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_qlog(transport: &mut quinn::TransportConfig, quic: &Resolved, role: &str) -> Result<()> {
22 let Some(dir) = quic.qlog_dir() else {
25 return Ok(());
26 };
27
28 #[cfg(feature = "qlog")]
29 {
30 let path = dir.join(format!("moq-{role}-{}.sqlog", std::process::id()));
32 let file = std::fs::File::create(&path).map_err(Error::CreateQlog)?;
33
34 let mut config = quinn::QlogConfig::default();
39 config.writer(Box::new(file)).title(Some(format!("moq-native {role}")));
40
41 transport.qlog_stream(config.into_stream());
42 tracing::info!(path = %path.display(), "writing qlog");
43 }
44
45 #[cfg(not(feature = "qlog"))]
46 let _ = (transport, dir, role);
47
48 Ok(())
49}
50
51fn apply_transport(transport: &mut quinn::TransportConfig, quic: &Resolved) {
53 transport.max_idle_timeout(Some(quic.idle_timeout.try_into().expect("idle timeout out of range")));
54 transport.keep_alive_interval(quic.keep_alive);
55
56 if !quic.mtu_discovery {
58 transport.mtu_discovery_config(None);
59 }
60
61 let max_streams = quinn::VarInt::from_u64(quic.max_streams).unwrap_or(quinn::VarInt::MAX);
62 transport.max_concurrent_bidi_streams(max_streams);
63 transport.max_concurrent_uni_streams(max_streams);
64
65 if let Some(gso) = quic.gso {
67 transport.enable_segmentation_offload(gso);
68 }
69
70 if let Some(family) = quic.congestion_control {
72 transport.congestion_controller_factory(congestion_factory(family));
73 }
74}
75
76fn congestion_factory(family: CongestionControl) -> Arc<dyn quinn::congestion::ControllerFactory + Send + Sync> {
78 match family {
79 CongestionControl::Loss => Arc::new(quinn::congestion::CubicConfig::default()),
80 CongestionControl::Delay => Arc::new(quinn::congestion::BbrConfig::default()),
81 }
82}
83
84#[derive(Debug, thiserror::Error)]
86#[non_exhaustive]
87pub enum Error {
88 #[error("failed to bind UDP socket")]
90 BindSocket(#[source] std::io::Error),
91
92 #[error("failed to create QUIC endpoint")]
94 CreateEndpoint(#[source] std::io::Error),
95
96 #[error("failed to create qlog file")]
98 CreateQlog(#[source] std::io::Error),
99
100 #[error("no async runtime")]
102 NoRuntime,
103
104 #[error("failed to get local address")]
106 LocalAddr(#[source] std::io::Error),
107
108 #[error("failed to resolve bind address")]
110 ResolveBind(#[source] std::io::Error),
111
112 #[error("invalid DNS name")]
114 InvalidDnsName,
115
116 #[error("failed DNS lookup")]
118 DnsLookup(#[source] std::io::Error),
119
120 #[error("no DNS entries")]
122 NoDnsEntries,
123
124 #[error("failed to fetch fingerprint")]
126 FetchFingerprint(#[source] reqwest::Error),
127
128 #[error("fingerprint request failed")]
130 FingerprintStatus(#[source] reqwest::Error),
131
132 #[error("failed to read fingerprint")]
134 ReadFingerprint(#[source] reqwest::Error),
135
136 #[error("invalid fingerprint")]
138 InvalidFingerprint(#[from] hex::FromHexError),
139
140 #[error("url scheme must be 'https', 'moqt', or 'moql'")]
142 InvalidScheme,
143
144 #[error("unsupported URL scheme: {0}")]
146 UnsupportedScheme(String),
147
148 #[error("missing handshake data")]
150 MissingHandshake,
151
152 #[error("missing ALPN")]
154 MissingAlpn,
155
156 #[error("failed to decode ALPN")]
158 DecodeAlpn(#[from] std::string::FromUtf8Error),
159
160 #[error("unsupported ALPN: {0}")]
162 UnsupportedAlpn(String),
163
164 #[error("missing server name for raw QUIC connection")]
166 MissingServerName,
167
168 #[error("failed to construct URL from server name")]
170 BuildUrl(#[source] url::ParseError),
171
172 #[error("quic_lb_nonce must be at least 4")]
174 QuicLbNonceTooSmall,
175
176 #[error("connection ID length ({0}) exceeds maximum of 20")]
178 QuicLbCidTooLong(usize),
179
180 #[error("failed to build client certificate verifier")]
182 ClientVerifier(#[source] rustls::server::VerifierBuilderError),
183
184 #[error(transparent)]
186 NoInitialCipherSuite(#[from] quinn::crypto::rustls::NoInitialCipherSuite),
187
188 #[error(transparent)]
190 Connect(#[from] quinn::ConnectError),
191
192 #[error(transparent)]
194 Connection(#[from] quinn::ConnectionError),
195
196 #[error(transparent)]
198 Client(#[from] web_transport_quinn::ClientError),
199
200 #[error(transparent)]
202 ConnectRejected(#[from] crate::ConnectError),
203
204 #[error(transparent)]
206 Server(#[from] web_transport_quinn::ServerError),
207
208 #[error("failed to establish QUIC connection")]
210 Establish(#[source] quinn::ConnectionError),
211
212 #[error("failed to receive WebTransport request")]
214 RecvRequest(#[source] web_transport_quinn::ServerError),
215
216 #[error(transparent)]
218 Tls(#[from] crate::tls::Error),
219}
220
221type Result<T> = std::result::Result<T, Error>;
222
223#[derive(Clone)]
226pub(crate) struct QuinnClient {
227 pub quic: quinn::Endpoint,
228 pub transport: Arc<quinn::TransportConfig>,
229 pub http_bootstrap: bool,
231 pub host_name: Option<String>,
233}
234
235impl QuinnClient {
236 pub fn new(config: &ClientConfig) -> Result<Self> {
237 let socket = crate::bind::udp(config.bind).map_err(Error::BindSocket)?;
238
239 let quic = config.quic.resolve();
240 let mut transport = quinn::TransportConfig::default();
241 apply_transport(&mut transport, &quic);
242 apply_qlog(&mut transport, &quic, "client")?;
243 let transport = Arc::new(transport);
244
245 let runtime = quinn::default_runtime().ok_or(Error::NoRuntime)?;
247 let endpoint_config = quinn::EndpointConfig::default();
248
249 let quic = quinn::Endpoint::new(endpoint_config, None, socket, runtime).map_err(Error::CreateEndpoint)?;
251
252 Ok(Self {
253 quic,
254 transport,
255 http_bootstrap: config.tls.allows_http_bootstrap(),
256 host_name: config.tls.host_name.clone(),
257 })
258 }
259
260 pub async fn connect(
261 &self,
262 tls: &rustls::ClientConfig,
263 url: Url,
264 versions: &moq_net::Versions,
265 ) -> Result<web_transport_quinn::Session> {
266 let mut url = url;
267 let mut config = tls.clone();
268
269 let host = url.host().ok_or(Error::InvalidDnsName)?.to_string();
270 let port = url.port().unwrap_or(443);
271
272 let local = self.quic.local_addr().map_err(Error::LocalAddr)?;
278 let addrs = tokio::net::lookup_host((host.clone(), port))
279 .await
280 .map_err(Error::DnsLookup)?;
281 let ip = crate::util::pick_addr(addrs, local).ok_or(Error::NoDnsEntries)?;
282
283 if url.scheme() == "http" {
284 if self.http_bootstrap {
288 let mut fingerprint = url.clone();
290 fingerprint.set_path("/certificate.sha256");
291 fingerprint.set_query(None);
292 fingerprint.set_fragment(None);
293
294 tracing::warn!(url = %fingerprint, "performing insecure HTTP request for certificate");
295
296 let resp = reqwest::get(fingerprint.as_str())
297 .await
298 .map_err(Error::FetchFingerprint)?
299 .error_for_status()
300 .map_err(Error::FingerprintStatus)?;
301
302 let fingerprint = resp.text().await.map_err(Error::ReadFingerprint)?;
303 let fingerprint = hex::decode(fingerprint.trim())?;
304
305 let verifier = FingerprintVerifier::new(config.crypto_provider().clone(), vec![fingerprint]);
306 config.dangerous().set_certificate_verifier(Arc::new(verifier));
307 } else {
308 tracing::warn!(
309 "ignoring insecure http:// fingerprint bootstrap; using the configured TLS verification"
310 );
311 }
312
313 url.set_scheme("https").expect("failed to set scheme");
314 }
315
316 let alpns: Vec<Vec<u8>> = match url.scheme() {
317 "https" => vec![web_transport_quinn::ALPN.as_bytes().to_vec()],
318 "moqt" | "moql" => versions.alpns().iter().map(|alpn| alpn.as_bytes().to_vec()).collect(),
319 _ => return Err(Error::InvalidScheme),
320 };
321
322 config.alpn_protocols = alpns;
323 config.key_log = Arc::new(rustls::KeyLogFile::new());
324
325 let config: quinn::crypto::rustls::QuicClientConfig = config.try_into()?;
326 let mut config = quinn::ClientConfig::new(Arc::new(config));
327 config.transport_config(self.transport.clone());
328
329 tracing::debug!(%url, %ip, "connecting");
330
331 let host_name = self.host_name.clone().unwrap_or(host);
333
334 let connection = self.quic.connect_with(config, ip, &host_name)?.await?;
335 tracing::Span::current().record("id", connection.stable_id());
336
337 let mut request = web_transport_quinn::proto::ConnectRequest::new(url.clone());
338 for alpn in versions.alpns() {
339 request = request.with_protocol(alpn.to_string());
340 }
341
342 let session = match url.scheme() {
343 "https" => web_transport_quinn::Session::connect(connection, request)
344 .await
345 .map_err(map_client_error)?,
346 "moqt" | "moql" => {
347 let handshake = connection
348 .handshake_data()
349 .ok_or(Error::MissingHandshake)?
350 .downcast::<quinn::crypto::rustls::HandshakeData>()
351 .unwrap();
352
353 let alpn = handshake.protocol.ok_or(Error::MissingAlpn)?;
354 let alpn = String::from_utf8(alpn)?;
355
356 let response = web_transport_quinn::proto::ConnectResponse::OK.with_protocol(alpn);
357 web_transport_quinn::Session::raw(connection, request, response)
358 }
359 _ => return Err(Error::UnsupportedScheme(url.scheme().to_string())),
360 };
361
362 Ok(session)
363 }
364}
365
366impl Error {
367 pub(crate) fn connect_error(&self) -> Option<crate::ConnectError> {
368 match self {
369 Self::ConnectRejected(err) => Some(*err),
370 Self::Client(err) => classify_client_error(err),
371 _ => None,
372 }
373 }
374}
375
376fn map_client_error(err: web_transport_quinn::ClientError) -> Error {
377 if let Some(err) = classify_client_error(&err) {
378 return err.into();
379 }
380
381 err.into()
382}
383
384fn classify_client_error(err: &web_transport_quinn::ClientError) -> Option<crate::ConnectError> {
385 match err {
386 web_transport_quinn::ClientError::HttpError(err) => classify_connect_error(err),
387 _ => None,
388 }
389}
390
391fn classify_connect_error(err: &web_transport_quinn::ConnectError) -> Option<crate::ConnectError> {
392 match err {
393 web_transport_quinn::ConnectError::ErrorStatus(status) => crate::ConnectError::from_status_u16(status.as_u16()),
394 web_transport_quinn::ConnectError::ProtoError(err) => classify_proto_error(err),
395 _ => None,
396 }
397}
398
399fn classify_proto_error(err: &web_transport_quinn::proto::ConnectError) -> Option<crate::ConnectError> {
400 match err {
401 web_transport_quinn::proto::ConnectError::ErrorStatus(status)
402 | web_transport_quinn::proto::ConnectError::WrongStatus(Some(status)) => {
403 crate::ConnectError::from_status_u16(status.as_u16())
404 }
405 _ => None,
406 }
407}
408
409pub(crate) struct QuinnServer {
412 pub quic: quinn::Endpoint,
413 pub certs: Arc<ServeCerts>,
414}
415
416impl QuinnServer {
417 pub fn new(config: ServerConfig) -> Result<Self> {
418 let quic = config.quic.resolve();
419 let mut transport = quinn::TransportConfig::default();
420 apply_transport(&mut transport, &quic);
421 apply_qlog(&mut transport, &quic, "server")?;
422 let transport = Arc::new(transport);
423
424 let provider = crate::crypto::provider();
425
426 let certs = ServeCerts::new(provider.clone());
427 certs.load_certs(&config.tls)?;
428 let certs = Arc::new(certs);
429
430 let tls_builder = rustls::ServerConfig::builder_with_provider(provider.clone())
431 .with_protocol_versions(&[&rustls::version::TLS13])
432 .map_err(crate::tls::Error::from)?;
433
434 let mut tls = if config.tls.root.is_empty() {
435 tls_builder.with_no_client_auth().with_cert_resolver(certs.clone())
436 } else {
437 let roots = config.tls.load_roots()?;
438 let verifier = rustls::server::WebPkiClientVerifier::builder_with_provider(Arc::new(roots), provider)
439 .allow_unauthenticated()
440 .build()
441 .map_err(Error::ClientVerifier)?;
442 tls_builder
443 .with_client_cert_verifier(verifier)
444 .with_cert_resolver(certs.clone())
445 };
446
447 let mut alpns: Vec<Vec<u8>> = config
449 .versions()
450 .alpns()
451 .iter()
452 .map(|alpn| alpn.as_bytes().to_vec())
453 .collect();
454 alpns.push(web_transport_quinn::ALPN.as_bytes().to_vec());
455
456 tls.alpn_protocols = alpns;
457 tls.key_log = Arc::new(rustls::KeyLogFile::new());
458
459 let tls: quinn::crypto::rustls::QuicServerConfig = tls.try_into()?;
460 let mut tls = quinn::ServerConfig::with_crypto(Arc::new(tls));
461 tls.transport_config(transport);
462
463 if let Some(addr) = config.quic.preferred_v4 {
466 tls.preferred_address_v4(Some(addr));
467 }
468 if let Some(addr) = config.quic.preferred_v6 {
469 tls.preferred_address_v6(Some(addr));
470 }
471
472 let runtime = quinn::default_runtime().ok_or(Error::NoRuntime)?;
474
475 let listen =
476 crate::util::resolve(config.bind.as_deref(), crate::server::DEFAULT_BIND).map_err(Error::ResolveBind)?;
477
478 let mut endpoint_config = quinn::EndpointConfig::default();
480 if let Some(server_id) = config.quic.quic_lb_id {
481 let nonce_len = config.quic.quic_lb_nonce.unwrap_or(8);
482 if nonce_len < 4 {
483 return Err(Error::QuicLbNonceTooSmall);
484 }
485
486 let cid_len = 1 + server_id.len() + nonce_len;
487 if cid_len > 20 {
488 return Err(Error::QuicLbCidTooLong(cid_len));
489 }
490
491 tracing::info!(
492 ?server_id,
493 nonce_len,
494 "using QUIC-LB compatible connection ID generation"
495 );
496 endpoint_config.cid_generator(move || Box::new(ServerIdGenerator::new(server_id.clone(), nonce_len)));
497 }
498
499 let socket = crate::bind::udp(listen).map_err(Error::BindSocket)?;
500
501 let quic = quinn::Endpoint::new(endpoint_config, Some(tls), socket, runtime).map_err(Error::CreateEndpoint)?;
503
504 tokio::spawn(crate::tls::reload_certs(certs.clone(), config.tls.clone()));
507
508 Ok(Self { quic, certs })
509 }
510
511 pub fn accept(&self) -> impl std::future::Future<Output = Option<quinn::Incoming>> + '_ {
512 self.quic.accept()
513 }
514
515 pub fn certificates(&self) -> crate::tls::Certificates {
516 crate::tls::Certificates::new(self.certs.info.clone())
517 }
518
519 pub fn local_addr(&self) -> Result<net::SocketAddr> {
520 self.quic.local_addr().map_err(Error::LocalAddr)
521 }
522
523 pub fn close(&self) {
524 self.quic.close(quinn::VarInt::from_u32(0), b"server shutdown");
525 }
526}
527
528pub(crate) async fn accept(
535 conn: quinn::Incoming,
536 alpns: Vec<&'static str>,
537) -> Result<(
538 web_transport_quinn::Session,
539 Option<Url>,
540 Option<crate::tls::PeerIdentity>,
541)> {
542 let mut conn = conn.accept()?;
543
544 let handshake = conn
545 .handshake_data()
546 .await?
547 .downcast::<quinn::crypto::rustls::HandshakeData>()
548 .unwrap();
549
550 let alpn = handshake.protocol.ok_or(Error::MissingAlpn)?;
551 let alpn = String::from_utf8(alpn)?;
552 let host = handshake.server_name.unwrap_or_default();
553
554 tracing::debug!(%host, ip = %conn.remote_address(), %alpn, "accepting");
555
556 let conn = conn.await.map_err(Error::Establish)?;
558
559 let span = tracing::Span::current();
560 span.record("id", conn.stable_id()); tracing::debug!(%host, ip = %conn.remote_address(), %alpn, "accepted");
562
563 match alpn.as_str() {
564 web_transport_quinn::ALPN => {
565 let request = web_transport_quinn::Request::accept(conn)
568 .await
569 .map_err(Error::RecvRequest)?;
570 let url = Some(request.url.clone());
571 let identity = crate::tls::PeerIdentity::from_any(request.conn().peer_identity());
572
573 let mut response = web_transport_quinn::proto::ConnectResponse::OK;
574 if let Some(protocol) = request.protocols.iter().find(|p| alpns.contains(&p.as_str())) {
580 response = response.with_protocol(protocol);
581 }
582 let session = request.respond(response).await.map_err(Error::Server)?;
583 Ok((session, url, identity))
584 }
585 alpn if alpns.contains(&alpn) => {
590 let host_str = if host.contains(':') {
596 format!("[{}]", host)
597 } else {
598 host.clone()
599 };
600 let url = format!("moqt://{}", host_str).parse::<Url>().map_err(Error::BuildUrl)?;
601 let request = web_transport_quinn::proto::ConnectRequest::new(url);
602 let response = web_transport_quinn::proto::ConnectResponse::OK.with_protocol(alpn);
603 let identity = crate::tls::PeerIdentity::from_any(conn.peer_identity());
604 let session = web_transport_quinn::Session::raw(conn, request, response);
606 Ok((session, None, identity))
607 }
608 _ => Err(Error::UnsupportedAlpn(alpn)),
609 }
610}
611
612struct ServerIdGenerator {
615 server_id: ServerId,
616 nonce_len: usize,
617}
618
619impl ServerIdGenerator {
620 fn new(server_id: ServerId, nonce_len: usize) -> Self {
621 Self { server_id, nonce_len }
622 }
623}
624
625impl quinn::ConnectionIdGenerator for ServerIdGenerator {
626 fn generate_cid(&mut self) -> quinn::ConnectionId {
627 use rand::RngExt;
628 let cid_len = self.cid_len();
629 let mut cid = Vec::with_capacity(cid_len);
630 cid.push((cid_len - 1) as u8);
632 cid.extend(self.server_id.0.iter());
633 cid.extend(rand::rng().random_iter::<u8>().take(self.nonce_len));
634 quinn::ConnectionId::new(cid.as_slice())
635 }
636
637 fn cid_len(&self) -> usize {
638 1 + self.server_id.len() + self.nonce_len
639 }
640
641 fn cid_lifetime(&self) -> Option<Duration> {
642 None
643 }
644}
645
646#[cfg(test)]
647mod tests {
648 use super::*;
649
650 #[test]
653 fn congestion_factory_maps_each_family() {
654 let now = std::time::Instant::now();
655 let mtu = 1200;
656
657 let loss = congestion_factory(CongestionControl::Loss).build(now, mtu);
658 assert!(loss.into_any().downcast::<quinn::congestion::Cubic>().is_ok());
659
660 let delay = congestion_factory(CongestionControl::Delay).build(now, mtu);
661 assert!(delay.into_any().downcast::<quinn::congestion::Bbr>().is_ok());
662 }
663
664 #[tokio::test]
667 async fn delay_reaches_the_live_connection() {
668 let server_config = ServerConfig {
669 bind: Some("127.0.0.1:0".to_string()),
670 tls: crate::tls::Server {
671 generate: vec!["localhost".into()],
672 ..Default::default()
673 },
674 quic: crate::quic::Server {
675 congestion_control: Some(CongestionControl::Delay),
676 ..Default::default()
677 },
678 ..Default::default()
679 };
680
681 let server = QuinnServer::new(server_config).expect("server init");
682 let addr = server.local_addr().expect("local addr");
683
684 let accepted = tokio::spawn(async move {
685 let incoming = server.accept().await.expect("no incoming connection");
686 let conn = incoming.accept().expect("accept").await.expect("handshake");
687 conn.congestion_state()
688 .into_any()
689 .downcast::<quinn::congestion::Bbr>()
690 .is_ok()
691 });
692
693 let mut tls_config = crate::tls::Client::default();
695 tls_config.disable_verify = Some(true);
696
697 let client_config = ClientConfig {
698 bind: "127.0.0.1:0".parse().unwrap(),
699 tls: tls_config,
700 quic: crate::quic::Client {
701 congestion_control: Some(CongestionControl::Delay),
702 ..Default::default()
703 },
704 ..Default::default()
705 };
706
707 let tls = client_config.tls.build().expect("tls config");
708 let client = QuinnClient::new(&client_config).expect("client init");
709 let url: Url = format!("moqt://127.0.0.1:{}", addr.port()).parse().unwrap();
711
712 tokio::time::timeout(Duration::from_secs(5), async move {
715 let session = client
716 .connect(&tls, url, &moq_net::Versions::default())
717 .await
718 .expect("connect failed");
719
720 assert!(
722 session
723 .congestion_state()
724 .into_any()
725 .downcast::<quinn::congestion::Bbr>()
726 .is_ok(),
727 "client connection is not running BBR"
728 );
729 assert!(
730 accepted.await.expect("server task panicked"),
731 "server connection is not running BBR"
732 );
733 })
734 .await
735 .expect("test timed out");
736 }
737}