1use crate::RedactedUrl;
4use crate::client::ClientConfig;
5use crate::quic::CongestionControl;
6use crate::quic::Resolved;
7use crate::quic::ServerId;
8use crate::server::ServerConfig;
9use crate::tls::{FingerprintVerifier, ServeCerts};
10use std::net;
11use std::sync::Arc;
12use std::time::Duration;
13use url::Url;
14
15pub use web_transport_quinn;
16
17fn apply_qlog(transport: &mut quinn::TransportConfig, quic: &Resolved, role: &str) -> Result<()> {
23 let Some(dir) = quic.qlog_dir() else {
26 return Ok(());
27 };
28
29 #[cfg(feature = "qlog")]
30 {
31 let path = dir.join(format!("moq-{role}-{}.sqlog", std::process::id()));
33 let file = std::fs::File::create(&path).map_err(Error::CreateQlog)?;
34
35 let mut config = quinn::QlogConfig::default();
40 config.writer(Box::new(file)).title(Some(format!("moq-native {role}")));
41
42 transport.qlog_stream(config.into_stream());
43 tracing::info!(path = %path.display(), "writing qlog");
44 }
45
46 #[cfg(not(feature = "qlog"))]
47 let _ = (transport, dir, role);
48
49 Ok(())
50}
51
52fn apply_transport(transport: &mut quinn::TransportConfig, quic: &Resolved) {
54 transport.max_idle_timeout(Some(quic.idle_timeout.try_into().expect("idle timeout out of range")));
55 transport.keep_alive_interval(quic.keep_alive);
56
57 if !quic.mtu_discovery {
59 transport.mtu_discovery_config(None);
60 }
61
62 let max_streams = quinn::VarInt::from_u64(quic.max_streams).unwrap_or(quinn::VarInt::MAX);
63 transport.max_concurrent_bidi_streams(max_streams);
64 transport.max_concurrent_uni_streams(max_streams);
65
66 if let Some(gso) = quic.gso {
68 transport.enable_segmentation_offload(gso);
69 }
70
71 transport.congestion_controller_factory(congestion_factory(quic.congestion()));
72}
73
74fn congestion_factory(family: CongestionControl) -> Arc<dyn quinn::congestion::ControllerFactory + Send + Sync> {
76 match family {
77 CongestionControl::Loss => Arc::new(quinn::congestion::CubicConfig::default()),
78 CongestionControl::Delay => Arc::new(quinn::congestion::BbrConfig::default()),
79 }
80}
81
82#[derive(Debug, thiserror::Error)]
84#[non_exhaustive]
85pub enum Error {
86 #[error("failed to bind UDP socket")]
88 BindSocket(#[source] std::io::Error),
89
90 #[error("failed to create QUIC endpoint")]
92 CreateEndpoint(#[source] std::io::Error),
93
94 #[error("failed to create qlog file")]
96 CreateQlog(#[source] std::io::Error),
97
98 #[error("no async runtime")]
100 NoRuntime,
101
102 #[error("failed to get local address")]
104 LocalAddr(#[source] std::io::Error),
105
106 #[error("failed to resolve bind address")]
108 ResolveBind(#[source] std::io::Error),
109
110 #[error("invalid DNS name")]
112 InvalidDnsName,
113
114 #[error("failed DNS lookup")]
116 DnsLookup(#[source] std::io::Error),
117
118 #[error("no DNS entries")]
120 NoDnsEntries,
121
122 #[error("failed to fetch fingerprint")]
124 FetchFingerprint(#[source] reqwest::Error),
125
126 #[error("fingerprint request failed")]
128 FingerprintStatus(#[source] reqwest::Error),
129
130 #[error("failed to read fingerprint")]
132 ReadFingerprint(#[source] reqwest::Error),
133
134 #[error("invalid fingerprint")]
136 InvalidFingerprint(#[from] hex::FromHexError),
137
138 #[error("url scheme must be 'https', 'moqt', or 'moql'")]
140 InvalidScheme,
141
142 #[error("unsupported URL scheme: {0}")]
144 UnsupportedScheme(String),
145
146 #[error("missing handshake data")]
148 MissingHandshake,
149
150 #[error("missing ALPN")]
152 MissingAlpn,
153
154 #[error("failed to decode ALPN")]
156 DecodeAlpn(#[from] std::string::FromUtf8Error),
157
158 #[error("unsupported ALPN: {0}")]
160 UnsupportedAlpn(String),
161
162 #[error("missing server name for raw QUIC connection")]
164 MissingServerName,
165
166 #[error("failed to construct URL from server name")]
168 BuildUrl(#[source] url::ParseError),
169
170 #[error("quic_lb_nonce must be at least 4")]
172 QuicLbNonceTooSmall,
173
174 #[error("connection ID length ({0}) exceeds maximum of 20")]
176 QuicLbCidTooLong(usize),
177
178 #[error("failed to build client certificate verifier")]
180 ClientVerifier(#[source] rustls::server::VerifierBuilderError),
181
182 #[error(transparent)]
184 NoInitialCipherSuite(#[from] quinn::crypto::rustls::NoInitialCipherSuite),
185
186 #[error(transparent)]
188 Connect(#[from] quinn::ConnectError),
189
190 #[error(transparent)]
192 Connection(#[from] quinn::ConnectionError),
193
194 #[error(transparent)]
196 Client(#[from] web_transport_quinn::ClientError),
197
198 #[error(transparent)]
200 ConnectRejected(#[from] crate::ConnectError),
201
202 #[error(transparent)]
204 Server(#[from] web_transport_quinn::ServerError),
205
206 #[error("failed to establish QUIC connection")]
208 Establish(#[source] quinn::ConnectionError),
209
210 #[error("failed to receive WebTransport request")]
212 RecvRequest(#[source] web_transport_quinn::ServerError),
213
214 #[error(transparent)]
216 Tls(#[from] crate::tls::Error),
217
218 #[error("all {} connection attempts failed: {}", .0.len(), crate::failover::describe(.0))]
224 Failover(Vec<crate::failover::Failure<Error>>),
225}
226
227impl crate::failover::Aggregate for Error {
228 fn aggregate(failures: Vec<crate::failover::Failure<Self>>) -> Self {
229 Self::Failover(failures)
230 }
231
232 fn resolve(error: Option<std::io::Error>) -> Self {
233 match error {
234 Some(error) => Self::DnsLookup(error),
235 None => Self::NoDnsEntries,
236 }
237 }
238}
239
240type Result<T> = std::result::Result<T, Error>;
241
242#[derive(Clone)]
245pub(crate) struct QuinnClient {
246 pub quic: quinn::Endpoint,
247 pub transport: Arc<quinn::TransportConfig>,
248 pub http_bootstrap: bool,
250 pub host_name: Option<String>,
252 pub failover_delay: Duration,
254 pub resolution_delay: Duration,
257 dual_stack: bool,
261}
262
263impl QuinnClient {
264 pub fn new(config: &ClientConfig) -> Result<Self> {
265 let socket = crate::bind::udp(config.bind).map_err(Error::BindSocket)?;
266 let dual_stack = crate::bind::udp_is_dual_stack(&socket);
267
268 let quic = config.quic.resolve();
269 let mut transport = quinn::TransportConfig::default();
270 apply_transport(&mut transport, &quic);
271 apply_qlog(&mut transport, &quic, "client")?;
272 let transport = Arc::new(transport);
273
274 let runtime = quinn::default_runtime().ok_or(Error::NoRuntime)?;
276 let endpoint_config = quinn::EndpointConfig::default();
277
278 let quic = quinn::Endpoint::new(endpoint_config, None, socket, runtime).map_err(Error::CreateEndpoint)?;
280
281 Ok(Self {
282 quic,
283 transport,
284 http_bootstrap: config.tls.allows_http_bootstrap(),
285 host_name: config.tls.host_name.clone(),
286 failover_delay: config.resolved_failover_delay(),
287 resolution_delay: config.resolved_resolution_delay(),
288 dual_stack,
289 })
290 }
291
292 pub async fn connect(
293 &self,
294 tls: &rustls::ClientConfig,
295 url: Url,
296 versions: &moq_net::Versions,
297 ) -> Result<web_transport_quinn::Session> {
298 let mut url = url;
299 let mut config = tls.clone();
300
301 let target = url.host().ok_or(Error::InvalidDnsName)?;
302 let host = target.to_string();
303 let port = url.port().unwrap_or(443);
304
305 let local = self.quic.local_addr().map_err(Error::LocalAddr)?;
309 let candidates =
310 crate::resolve::Candidates::resolve(target, port, self.resolution_delay).with_local(local, self.dual_stack);
311
312 if url.scheme() == "http" {
313 if self.http_bootstrap {
317 let mut fingerprint = url.clone();
319 fingerprint.set_path("/certificate.sha256");
320 fingerprint.set_query(None);
321 fingerprint.set_fragment(None);
322
323 tracing::warn!(url = %RedactedUrl::new(&fingerprint), "performing insecure HTTP request for certificate");
324
325 let resp = reqwest::get(fingerprint.as_str())
326 .await
327 .map_err(Error::FetchFingerprint)?
328 .error_for_status()
329 .map_err(Error::FingerprintStatus)?;
330
331 let fingerprint = resp.text().await.map_err(Error::ReadFingerprint)?;
332 let fingerprint = hex::decode(fingerprint.trim())?;
333
334 let verifier = FingerprintVerifier::new(config.crypto_provider().clone(), vec![fingerprint]);
335 config.dangerous().set_certificate_verifier(Arc::new(verifier));
336 } else {
337 tracing::warn!(
338 "ignoring insecure http:// fingerprint bootstrap; using the configured TLS verification"
339 );
340 }
341
342 url.set_scheme("https").expect("failed to set scheme");
343 }
344
345 let alpns: Vec<Vec<u8>> = match url.scheme() {
346 "https" => vec![web_transport_quinn::ALPN.as_bytes().to_vec()],
347 "moqt" | "moql" => versions.alpns().iter().map(|alpn| alpn.as_bytes().to_vec()).collect(),
348 _ => return Err(Error::InvalidScheme),
349 };
350
351 config.alpn_protocols = alpns;
352 config.key_log = Arc::new(rustls::KeyLogFile::new());
353
354 let config: quinn::crypto::rustls::QuicClientConfig = config.try_into()?;
355 let mut config = quinn::ClientConfig::new(Arc::new(config));
356 config.transport_config(self.transport.clone());
357
358 tracing::debug!(url = %RedactedUrl::new(&url), "connecting");
359
360 let host_name = self.host_name.clone().unwrap_or(host);
362
363 let connection = crate::failover::race(candidates, self.failover_delay, |addr| {
367 let endpoint = self.quic.clone();
368 let config = config.clone();
369 let host_name = host_name.clone();
370 async move { Ok::<_, Error>(endpoint.connect_with(config, addr, &host_name)?.await?) }
371 })
372 .await?;
373 tracing::Span::current().record("id", connection.stable_id());
374
375 let session = match url.scheme() {
376 "https" => {
377 let mut request = web_transport_quinn::proto::ConnectRequest::new(url.clone());
378 for alpn in versions.alpns() {
379 request = request.with_protocol(alpn.to_string());
380 }
381 web_transport_quinn::Session::connect(connection, request)
382 .await
383 .map_err(map_client_error)?
384 }
385 "moqt" | "moql" => web_transport_quinn::Session::raw(connection),
386 _ => return Err(Error::UnsupportedScheme(url.scheme().to_string())),
387 };
388
389 Ok(session)
390 }
391}
392
393impl Error {
394 pub(crate) fn connect_error(&self) -> Option<crate::ConnectError> {
395 match self {
396 Self::ConnectRejected(err) => Some(*err),
397 Self::Client(err) => classify_client_error(err),
398 Self::Failover(failures) => failures.iter().find_map(|failure| failure.error.connect_error()),
399 _ => None,
400 }
401 }
402
403 pub(crate) fn status(&self) -> Option<u16> {
408 match self {
409 Self::FetchFingerprint(err) | Self::FingerprintStatus(err) | Self::ReadFingerprint(err) => {
410 err.status().map(|status| status.as_u16())
411 }
412 Self::Client(err) => client_status(err),
413 Self::Failover(failures) => {
417 let mut settled = None;
418 for failure in failures {
419 match failure.error.status() {
420 Some(status) if !crate::error::status_retryable(status) => settled = Some(status),
421 _ => return None,
422 }
423 }
424 settled
425 }
426 _ => None,
427 }
428 }
429}
430
431fn map_client_error(err: web_transport_quinn::ClientError) -> Error {
432 if let Some(err) = classify_client_error(&err) {
433 return err.into();
434 }
435
436 err.into()
437}
438
439fn classify_client_error(err: &web_transport_quinn::ClientError) -> Option<crate::ConnectError> {
440 client_status(err).and_then(crate::ConnectError::from_status_u16)
441}
442
443fn client_status(err: &web_transport_quinn::ClientError) -> Option<u16> {
451 match err {
452 web_transport_quinn::ClientError::HttpError(err) => connect_status(err),
453 _ => None,
454 }
455}
456
457fn connect_status(err: &web_transport_quinn::ConnectError) -> Option<u16> {
458 match err {
459 web_transport_quinn::ConnectError::ErrorStatus(status) => Some(status.as_u16()),
460 web_transport_quinn::ConnectError::ProtoError(err) => proto_status(err),
461 _ => None,
462 }
463}
464
465fn proto_status(err: &web_transport_quinn::proto::ConnectError) -> Option<u16> {
466 match err {
467 web_transport_quinn::proto::ConnectError::ErrorStatus(status)
468 | web_transport_quinn::proto::ConnectError::WrongStatus(Some(status)) => Some(status.as_u16()),
469 _ => None,
470 }
471}
472
473pub(crate) struct QuinnServer {
476 pub quic: quinn::Endpoint,
477 pub certs: Arc<ServeCerts>,
478}
479
480impl QuinnServer {
481 pub fn new(config: ServerConfig) -> Result<Self> {
482 let quic = config.quic.resolve();
483 let mut transport = quinn::TransportConfig::default();
484 apply_transport(&mut transport, &quic);
485 apply_qlog(&mut transport, &quic, "server")?;
486 let transport = Arc::new(transport);
487
488 let provider = crate::crypto::provider();
489
490 let certs = ServeCerts::new(provider.clone());
491 certs.load_certs(&config.tls)?;
492 let certs = Arc::new(certs);
493
494 let tls_builder = rustls::ServerConfig::builder_with_provider(provider.clone())
495 .with_protocol_versions(&[&rustls::version::TLS13])
496 .map_err(crate::tls::Error::from)?;
497
498 let mut tls = if config.tls.root.is_empty() {
499 tls_builder.with_no_client_auth().with_cert_resolver(certs.clone())
500 } else {
501 let verifier = config.tls.client_verifier(provider)?;
502 tls_builder
503 .with_client_cert_verifier(verifier)
504 .with_cert_resolver(certs.clone())
505 };
506
507 let mut alpns: Vec<Vec<u8>> = config
509 .versions()
510 .alpns()
511 .iter()
512 .map(|alpn| alpn.as_bytes().to_vec())
513 .collect();
514 alpns.push(web_transport_quinn::ALPN.as_bytes().to_vec());
515
516 tls.alpn_protocols = alpns;
517 tls.key_log = Arc::new(rustls::KeyLogFile::new());
518 config.tls.disable_resumption(&mut tls);
519
520 let tls: quinn::crypto::rustls::QuicServerConfig = tls.try_into()?;
521 let mut tls = quinn::ServerConfig::with_crypto(Arc::new(tls));
522 tls.transport_config(transport);
523
524 if let Some(addr) = config.quic.preferred_v4 {
527 tls.preferred_address_v4(Some(addr));
528 }
529 if let Some(addr) = config.quic.preferred_v6 {
530 tls.preferred_address_v6(Some(addr));
531 }
532
533 let runtime = quinn::default_runtime().ok_or(Error::NoRuntime)?;
535
536 let listen =
537 crate::util::resolve(config.bind.as_deref(), crate::server::DEFAULT_BIND).map_err(Error::ResolveBind)?;
538
539 let mut endpoint_config = quinn::EndpointConfig::default();
541 if let Some(server_id) = config.quic.quic_lb_id {
542 let nonce_len = config.quic.quic_lb_nonce.unwrap_or(8);
543 if nonce_len < 4 {
544 return Err(Error::QuicLbNonceTooSmall);
545 }
546
547 let cid_len = 1 + server_id.len() + nonce_len;
548 if cid_len > 20 {
549 return Err(Error::QuicLbCidTooLong(cid_len));
550 }
551
552 tracing::info!(
553 ?server_id,
554 nonce_len,
555 "using QUIC-LB compatible connection ID generation"
556 );
557 endpoint_config.cid_generator(move || Box::new(ServerIdGenerator::new(server_id.clone(), nonce_len)));
558 }
559
560 let socket = crate::bind::udp(listen).map_err(Error::BindSocket)?;
561
562 let quic = quinn::Endpoint::new(endpoint_config, Some(tls), socket, runtime).map_err(Error::CreateEndpoint)?;
564
565 tokio::spawn(crate::tls::reload_certs(certs.clone(), config.tls.clone()));
568
569 Ok(Self { quic, certs })
570 }
571
572 pub fn accept(&self) -> impl std::future::Future<Output = Option<quinn::Incoming>> + '_ {
573 self.quic.accept()
574 }
575
576 pub fn certificates(&self) -> crate::tls::Certificates {
577 crate::tls::Certificates::new(self.certs.info.clone())
578 }
579
580 pub fn local_addr(&self) -> Result<net::SocketAddr> {
581 self.quic.local_addr().map_err(Error::LocalAddr)
582 }
583
584 pub fn close(&self) {
585 self.quic.close(quinn::VarInt::from_u32(0), b"server shutdown");
586 }
587}
588
589pub(crate) async fn accept(
596 conn: quinn::Incoming,
597 alpns: Vec<&'static str>,
598) -> Result<(
599 web_transport_quinn::Session,
600 Option<Url>,
601 Option<crate::tls::PeerIdentity>,
602)> {
603 let mut conn = conn.accept()?;
604
605 let handshake = conn
606 .handshake_data()
607 .await?
608 .downcast::<quinn::crypto::rustls::HandshakeData>()
609 .unwrap();
610
611 let alpn = handshake.protocol.ok_or(Error::MissingAlpn)?;
612 let alpn = String::from_utf8(alpn)?;
613 let host = handshake.server_name.unwrap_or_default();
614
615 tracing::debug!(%host, ip = %conn.remote_address(), %alpn, "accepting");
616
617 let conn = conn.await.map_err(Error::Establish)?;
619
620 let span = tracing::Span::current();
621 span.record("id", conn.stable_id()); tracing::debug!(%host, ip = %conn.remote_address(), %alpn, "accepted");
623
624 match alpn.as_str() {
625 web_transport_quinn::ALPN => {
626 let request = web_transport_quinn::Request::accept(conn)
629 .await
630 .map_err(Error::RecvRequest)?;
631 let url = Some(request.url.clone());
632 let identity = crate::tls::PeerIdentity::from_any(request.conn().peer_identity());
633
634 let mut response = web_transport_quinn::proto::ConnectResponse::OK;
635 if let Some(protocol) = request.protocols.iter().find(|p| alpns.contains(&p.as_str())) {
641 response = response.with_protocol(protocol);
642 }
643 let session = request.respond(response).await.map_err(Error::Server)?;
644 Ok((session, url, identity))
645 }
646 alpn if alpns.contains(&alpn) => {
651 let identity = crate::tls::PeerIdentity::from_any(conn.peer_identity());
652 let session = web_transport_quinn::Session::raw(conn);
654 Ok((session, None, identity))
655 }
656 _ => Err(Error::UnsupportedAlpn(alpn)),
657 }
658}
659
660struct ServerIdGenerator {
663 server_id: ServerId,
664 nonce_len: usize,
665}
666
667impl ServerIdGenerator {
668 fn new(server_id: ServerId, nonce_len: usize) -> Self {
669 Self { server_id, nonce_len }
670 }
671}
672
673impl quinn::ConnectionIdGenerator for ServerIdGenerator {
674 fn generate_cid(&mut self) -> quinn::ConnectionId {
675 use rand::RngExt;
676 let cid_len = self.cid_len();
677 let mut cid = Vec::with_capacity(cid_len);
678 cid.push((cid_len - 1) as u8);
680 cid.extend(self.server_id.0.iter());
681 cid.extend(rand::rng().random_iter::<u8>().take(self.nonce_len));
682 quinn::ConnectionId::new(cid.as_slice())
683 }
684
685 fn cid_len(&self) -> usize {
686 1 + self.server_id.len() + self.nonce_len
687 }
688
689 fn cid_lifetime(&self) -> Option<Duration> {
690 None
691 }
692}
693
694#[cfg(test)]
695mod tests {
696 use super::*;
697
698 fn connect_rejected(status: u16) -> Error {
699 Error::Client(web_transport_quinn::ClientError::HttpError(
700 web_transport_quinn::ConnectError::ErrorStatus(
701 web_transport_quinn::http::StatusCode::from_u16(status).unwrap(),
702 ),
703 ))
704 }
705
706 #[test]
710 fn a_rejected_connect_reports_its_status() {
711 for status in [400, 404, 405, 410, 501] {
712 assert_eq!(connect_rejected(status).status(), Some(status));
713 assert!(
714 !crate::error::status_retryable(status),
715 "{status} should stop the reconnect loop"
716 );
717 }
718
719 for status in [408, 429, 502, 503, 504] {
720 assert_eq!(connect_rejected(status).status(), Some(status));
721 assert!(crate::error::status_retryable(status), "{status} should be retried");
722 }
723
724 assert_eq!(
726 connect_rejected(401).connect_error(),
727 Some(crate::ConnectError::Unauthorized)
728 );
729 assert_eq!(
730 connect_rejected(403).connect_error(),
731 Some(crate::ConnectError::Forbidden)
732 );
733 }
734
735 #[test]
738 fn congestion_factory_maps_each_family() {
739 let now = std::time::Instant::now();
740 let mtu = 1200;
741
742 let loss = congestion_factory(CongestionControl::Loss).build(now, mtu);
743 assert!(loss.into_any().downcast::<quinn::congestion::Cubic>().is_ok());
744
745 let delay = congestion_factory(CongestionControl::Delay).build(now, mtu);
746 assert!(delay.into_any().downcast::<quinn::congestion::Bbr>().is_ok());
747 }
748
749 #[tokio::test]
752 async fn delay_reaches_the_live_connection() {
753 let server_config = ServerConfig {
754 bind: Some("127.0.0.1:0".to_string()),
755 tls: crate::tls::Server {
756 generate: vec!["localhost".into()],
757 ..Default::default()
758 },
759 quic: crate::quic::Server {
760 congestion_control: Some(CongestionControl::Delay),
761 ..Default::default()
762 },
763 ..Default::default()
764 };
765
766 let server = QuinnServer::new(server_config).expect("server init");
767 let addr = server.local_addr().expect("local addr");
768
769 let accepted = tokio::spawn(async move {
770 let incoming = server.accept().await.expect("no incoming connection");
771 let conn = incoming.accept().expect("accept").await.expect("handshake");
772 conn.congestion_state()
773 .into_any()
774 .downcast::<quinn::congestion::Bbr>()
775 .is_ok()
776 });
777
778 let mut tls_config = crate::tls::Client::default();
780 tls_config.disable_verify = Some(true);
781
782 let client_config = ClientConfig {
783 bind: "127.0.0.1:0".parse().unwrap(),
784 tls: tls_config,
785 quic: crate::quic::Client {
786 congestion_control: Some(CongestionControl::Delay),
787 ..Default::default()
788 },
789 ..Default::default()
790 };
791
792 let tls = client_config.tls.build().expect("tls config");
793 let client = QuinnClient::new(&client_config).expect("client init");
794 let url: Url = format!("moqt://127.0.0.1:{}", addr.port()).parse().unwrap();
796
797 tokio::time::timeout(Duration::from_secs(5), async move {
800 let session = client
801 .connect(&tls, url, &moq_net::Versions::default())
802 .await
803 .expect("connect failed");
804
805 assert!(
807 session
808 .congestion_state()
809 .into_any()
810 .downcast::<quinn::congestion::Bbr>()
811 .is_ok(),
812 "client connection is not running BBR"
813 );
814 assert!(
815 accepted.await.expect("server task panicked"),
816 "server connection is not running BBR"
817 );
818 })
819 .await
820 .expect("test timed out");
821 }
822}