1use std::net;
2use std::path::PathBuf;
3
4use crate::QuicBackend;
5use moq_lite::Session;
6use std::sync::{Arc, RwLock};
7use url::Url;
8#[cfg(feature = "iroh")]
9use web_transport_iroh::iroh;
10
11use anyhow::Context;
12
13use futures::FutureExt;
14use futures::future::BoxFuture;
15use futures::stream::FuturesUnordered;
16use futures::stream::StreamExt;
17
18#[serde_with::serde_as]
25#[derive(clap::Args, Clone, Default, Debug, serde::Serialize, serde::Deserialize)]
26#[serde(deny_unknown_fields)]
27#[non_exhaustive]
28pub struct ServerTlsConfig {
29 #[arg(long = "tls-cert", id = "tls-cert", env = "MOQ_SERVER_TLS_CERT")]
31 #[serde(default, skip_serializing_if = "Vec::is_empty")]
32 #[serde_as(as = "serde_with::OneOrMany<_>")]
33 pub cert: Vec<PathBuf>,
34
35 #[arg(long = "tls-key", id = "tls-key", env = "MOQ_SERVER_TLS_KEY")]
37 #[serde(default, skip_serializing_if = "Vec::is_empty")]
38 #[serde_as(as = "serde_with::OneOrMany<_>")]
39 pub key: Vec<PathBuf>,
40
41 #[arg(
44 long = "tls-generate",
45 id = "tls-generate",
46 value_delimiter = ',',
47 env = "MOQ_SERVER_TLS_GENERATE"
48 )]
49 #[serde(default, skip_serializing_if = "Vec::is_empty")]
50 #[serde_as(as = "serde_with::OneOrMany<_>")]
51 pub generate: Vec<String>,
52
53 #[arg(
62 long = "server-tls-root",
63 id = "server-tls-root",
64 value_delimiter = ',',
65 env = "MOQ_SERVER_TLS_ROOT"
66 )]
67 #[serde(default, skip_serializing_if = "Vec::is_empty")]
68 #[serde_as(as = "serde_with::OneOrMany<_>")]
69 pub root: Vec<PathBuf>,
70}
71
72impl ServerTlsConfig {
73 pub fn load_roots(&self) -> anyhow::Result<rustls::RootCertStore> {
75 use rustls::pki_types::CertificateDer;
76
77 let mut roots = rustls::RootCertStore::empty();
78 for path in &self.root {
79 let file = std::fs::File::open(path).context("failed to open root CA")?;
80 let mut reader = std::io::BufReader::new(file);
81 let certs: Vec<CertificateDer<'static>> = rustls_pemfile::certs(&mut reader)
82 .collect::<Result<_, _>>()
83 .context("failed to parse root CA PEM")?;
84 anyhow::ensure!(!certs.is_empty(), "no certificates found in root CA");
85 for cert in certs {
86 roots.add(cert).context("failed to add root CA")?;
87 }
88 }
89 Ok(roots)
90 }
91}
92
93#[derive(clap::Args, Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
95#[serde(deny_unknown_fields, default)]
96#[non_exhaustive]
97pub struct ServerConfig {
98 #[serde(alias = "listen")]
106 #[arg(id = "server-bind", long = "server-bind", alias = "listen", env = "MOQ_SERVER_BIND")]
107 pub bind: Option<String>,
108
109 #[arg(id = "server-backend", long = "server-backend", env = "MOQ_SERVER_BACKEND")]
112 pub backend: Option<QuicBackend>,
113
114 #[arg(id = "server-quic-lb-id", long = "server-quic-lb-id", env = "MOQ_SERVER_QUIC_LB_ID")]
117 #[serde(default, skip_serializing_if = "Option::is_none")]
118 pub quic_lb_id: Option<ServerId>,
119
120 #[arg(
123 id = "server-quic-lb-nonce",
124 long = "server-quic-lb-nonce",
125 requires = "server-quic-lb-id",
126 env = "MOQ_SERVER_QUIC_LB_NONCE"
127 )]
128 #[serde(default, skip_serializing_if = "Option::is_none")]
129 pub quic_lb_nonce: Option<usize>,
130
131 #[serde(skip_serializing_if = "Option::is_none")]
133 #[arg(
134 id = "server-max-streams",
135 long = "server-max-streams",
136 env = "MOQ_SERVER_MAX_STREAMS"
137 )]
138 pub max_streams: Option<u64>,
139
140 #[serde(default, skip_serializing_if = "Vec::is_empty")]
148 #[arg(id = "server-version", long = "server-version", env = "MOQ_SERVER_VERSION")]
149 pub version: Vec<moq_lite::Version>,
150
151 #[command(flatten)]
152 #[serde(default)]
153 pub tls: ServerTlsConfig,
154}
155
156impl ServerConfig {
157 pub fn init(self) -> anyhow::Result<Server> {
158 Server::new(self)
159 }
160
161 pub fn versions(&self) -> moq_lite::Versions {
163 if self.version.is_empty() {
164 moq_lite::Versions::all()
165 } else {
166 moq_lite::Versions::from(self.version.clone())
167 }
168 }
169}
170
171pub(crate) const DEFAULT_BIND: &str = "[::]:443";
173
174pub struct Server {
178 moq: moq_lite::Server,
179 versions: moq_lite::Versions,
180 accept: FuturesUnordered<BoxFuture<'static, anyhow::Result<Request>>>,
181 #[cfg(feature = "iroh")]
182 iroh: Option<iroh::Endpoint>,
183 #[cfg(feature = "noq")]
184 noq: Option<crate::noq::NoqServer>,
185 #[cfg(feature = "quinn")]
186 quinn: Option<crate::quinn::QuinnServer>,
187 #[cfg(feature = "quiche")]
188 quiche: Option<crate::quiche::QuicheServer>,
189 #[cfg(feature = "websocket")]
190 websocket: Option<crate::websocket::WebSocketListener>,
191}
192
193impl Server {
194 pub fn new(config: ServerConfig) -> anyhow::Result<Self> {
195 let backend = config.backend.clone().unwrap_or({
196 #[cfg(feature = "quinn")]
197 {
198 QuicBackend::Quinn
199 }
200 #[cfg(all(feature = "noq", not(feature = "quinn")))]
201 {
202 QuicBackend::Noq
203 }
204 #[cfg(all(feature = "quiche", not(feature = "quinn"), not(feature = "noq")))]
205 {
206 QuicBackend::Quiche
207 }
208 #[cfg(all(not(feature = "quiche"), not(feature = "quinn"), not(feature = "noq")))]
209 panic!("no QUIC backend compiled; enable noq, quinn, or quiche feature");
210 });
211
212 let versions = config.versions();
213
214 if !config.tls.root.is_empty() {
215 #[cfg(feature = "quinn")]
216 let quinn_backend = matches!(backend, QuicBackend::Quinn);
217 #[cfg(not(feature = "quinn"))]
218 let quinn_backend = false;
219 anyhow::ensure!(quinn_backend, "tls.root (mTLS) is only supported by the quinn backend");
220 }
221
222 #[cfg(feature = "noq")]
223 #[allow(unreachable_patterns)]
224 let noq = match backend {
225 QuicBackend::Noq => Some(crate::noq::NoqServer::new(config.clone())?),
226 _ => None,
227 };
228
229 #[cfg(feature = "quinn")]
230 #[allow(unreachable_patterns)]
231 let quinn = match backend {
232 QuicBackend::Quinn => Some(crate::quinn::QuinnServer::new(config.clone())?),
233 _ => None,
234 };
235
236 #[cfg(feature = "quiche")]
237 let quiche = match backend {
238 QuicBackend::Quiche => Some(crate::quiche::QuicheServer::new(config)?),
239 _ => None,
240 };
241
242 Ok(Server {
243 accept: Default::default(),
244 moq: moq_lite::Server::new().with_versions(versions.clone()),
245 versions,
246 #[cfg(feature = "iroh")]
247 iroh: None,
248 #[cfg(feature = "noq")]
249 noq,
250 #[cfg(feature = "quinn")]
251 quinn,
252 #[cfg(feature = "quiche")]
253 quiche,
254 #[cfg(feature = "websocket")]
255 websocket: None,
256 })
257 }
258
259 #[cfg(feature = "websocket")]
265 pub fn with_websocket(mut self, websocket: Option<crate::websocket::WebSocketListener>) -> Self {
266 self.websocket = websocket;
267 self
268 }
269
270 #[cfg(feature = "iroh")]
271 pub fn with_iroh(mut self, iroh: Option<iroh::Endpoint>) -> Self {
272 self.iroh = iroh;
273 self
274 }
275
276 pub fn with_publish(mut self, publish: impl Into<Option<moq_lite::OriginConsumer>>) -> Self {
277 self.moq = self.moq.with_publish(publish);
278 self
279 }
280
281 pub fn with_consume(mut self, consume: impl Into<Option<moq_lite::OriginProducer>>) -> Self {
282 self.moq = self.moq.with_consume(consume);
283 self
284 }
285
286 pub fn tls_info(&self) -> Arc<RwLock<ServerTlsInfo>> {
288 #[cfg(feature = "noq")]
289 if let Some(noq) = self.noq.as_ref() {
290 return noq.tls_info();
291 }
292 #[cfg(feature = "quinn")]
293 if let Some(quinn) = self.quinn.as_ref() {
294 return quinn.tls_info();
295 }
296 #[cfg(feature = "quiche")]
297 if let Some(quiche) = self.quiche.as_ref() {
298 return quiche.tls_info();
299 }
300 unreachable!("no QUIC backend compiled");
301 }
302
303 #[cfg(not(any(feature = "noq", feature = "quinn", feature = "quiche", feature = "iroh")))]
304 pub async fn accept(&mut self) -> Option<Request> {
305 unreachable!("no QUIC backend compiled; enable noq, quinn, quiche, or iroh feature");
306 }
307
308 #[cfg(any(feature = "noq", feature = "quinn", feature = "quiche", feature = "iroh"))]
316 pub async fn accept(&mut self) -> Option<Request> {
317 loop {
318 #[cfg(feature = "noq")]
320 let noq_accept = async {
321 #[cfg(feature = "noq")]
322 if let Some(noq) = self.noq.as_mut() {
323 return noq.accept().await;
324 }
325 None
326 };
327 #[cfg(not(feature = "noq"))]
328 let noq_accept = async { None::<()> };
329
330 #[cfg(feature = "iroh")]
331 let iroh_accept = async {
332 #[cfg(feature = "iroh")]
333 if let Some(endpoint) = self.iroh.as_mut() {
334 return endpoint.accept().await;
335 }
336 None
337 };
338 #[cfg(not(feature = "iroh"))]
339 let iroh_accept = async { None::<()> };
340
341 #[cfg(feature = "quinn")]
342 let quinn_accept = async {
343 #[cfg(feature = "quinn")]
344 if let Some(quinn) = self.quinn.as_mut() {
345 return quinn.accept().await;
346 }
347 None
348 };
349 #[cfg(not(feature = "quinn"))]
350 let quinn_accept = async { None::<()> };
351
352 #[cfg(feature = "quiche")]
353 let quiche_accept = async {
354 #[cfg(feature = "quiche")]
355 if let Some(quiche) = self.quiche.as_mut() {
356 return quiche.accept().await;
357 }
358 None
359 };
360 #[cfg(not(feature = "quiche"))]
361 let quiche_accept = async { None::<()> };
362
363 #[cfg(feature = "websocket")]
364 let ws_ref = self.websocket.as_ref();
365 #[cfg(feature = "websocket")]
366 let ws_accept = async {
367 match ws_ref {
368 Some(ws) => ws.accept().await,
369 None => std::future::pending().await,
370 }
371 };
372 #[cfg(not(feature = "websocket"))]
373 let ws_accept = std::future::pending::<Option<anyhow::Result<()>>>();
374
375 let server = self.moq.clone();
376 let versions = self.versions.clone();
377
378 tokio::select! {
379 Some(_conn) = noq_accept => {
380 #[cfg(feature = "noq")]
381 {
382 let alpns = versions.alpns();
383 self.accept.push(async move {
384 let noq = super::noq::NoqRequest::accept(_conn, alpns).await?;
385 Ok(Request {
386 server,
387 kind: RequestKind::Noq(noq),
388 })
389 }.boxed());
390 }
391 }
392 Some(_conn) = quinn_accept => {
393 #[cfg(feature = "quinn")]
394 {
395 let alpns = versions.alpns();
396 self.accept.push(async move {
397 let quinn = super::quinn::QuinnRequest::accept(_conn, alpns).await?;
398 Ok(Request {
399 server,
400 kind: RequestKind::Quinn(Box::new(quinn)),
401 })
402 }.boxed());
403 }
404 }
405 Some(_conn) = quiche_accept => {
406 #[cfg(feature = "quiche")]
407 {
408 let alpns = versions.alpns();
409 self.accept.push(async move {
410 let quiche = super::quiche::QuicheRequest::accept(_conn, alpns).await?;
411 Ok(Request {
412 server,
413 kind: RequestKind::Quiche(quiche),
414 })
415 }.boxed());
416 }
417 }
418 Some(_conn) = iroh_accept => {
419 #[cfg(feature = "iroh")]
420 self.accept.push(async move {
421 let iroh = super::iroh::IrohRequest::accept(_conn).await?;
422 Ok(Request {
423 server,
424 kind: RequestKind::Iroh(iroh),
425 })
426 }.boxed());
427 }
428 Some(_res) = ws_accept => {
429 #[cfg(feature = "websocket")]
430 match _res {
431 Ok(session) => {
432 return Some(Request {
433 server,
434 kind: RequestKind::WebSocket(session),
435 });
436 }
437 Err(err) => tracing::debug!(%err, "failed to accept WebSocket session"),
438 }
439 }
440 Some(res) = self.accept.next() => {
441 match res {
442 Ok(session) => return Some(session),
443 Err(err) => tracing::debug!(%err, "failed to accept session"),
444 }
445 }
446 _ = tokio::signal::ctrl_c() => {
447 self.close().await;
448 return None;
449 }
450 }
451 }
452 }
453
454 #[cfg(feature = "iroh")]
455 pub fn iroh_endpoint(&self) -> Option<&iroh::Endpoint> {
456 self.iroh.as_ref()
457 }
458
459 pub fn local_addr(&self) -> anyhow::Result<net::SocketAddr> {
460 #[cfg(feature = "noq")]
461 if let Some(noq) = self.noq.as_ref() {
462 return noq.local_addr();
463 }
464 #[cfg(feature = "quinn")]
465 if let Some(quinn) = self.quinn.as_ref() {
466 return quinn.local_addr();
467 }
468 #[cfg(feature = "quiche")]
469 if let Some(quiche) = self.quiche.as_ref() {
470 return quiche.local_addr();
471 }
472 unreachable!("no QUIC backend compiled");
473 }
474
475 #[cfg(feature = "websocket")]
476 pub fn websocket_local_addr(&self) -> Option<net::SocketAddr> {
477 self.websocket.as_ref().and_then(|ws| ws.local_addr().ok())
478 }
479
480 pub async fn close(&mut self) {
481 #[cfg(feature = "noq")]
482 if let Some(noq) = self.noq.as_mut() {
483 noq.close();
484 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
485 }
486 #[cfg(feature = "quinn")]
487 if let Some(quinn) = self.quinn.as_mut() {
488 quinn.close();
489 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
490 }
491 #[cfg(feature = "quiche")]
492 if let Some(quiche) = self.quiche.as_mut() {
493 quiche.close();
494 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
495 }
496 #[cfg(feature = "iroh")]
497 if let Some(iroh) = self.iroh.take() {
498 iroh.close().await;
499 }
500 #[cfg(feature = "websocket")]
501 {
502 let _ = self.websocket.take();
503 }
504 #[cfg(not(any(feature = "noq", feature = "quinn", feature = "quiche", feature = "iroh")))]
505 unreachable!("no QUIC backend compiled");
506 }
507}
508
509#[derive(Clone, Debug, Default)]
512#[non_exhaustive]
513pub struct PeerIdentity {}
514
515pub(crate) enum RequestKind {
517 #[cfg(feature = "noq")]
518 Noq(crate::noq::NoqRequest),
519 #[cfg(feature = "quinn")]
520 Quinn(Box<crate::quinn::QuinnRequest>),
521 #[cfg(feature = "quiche")]
522 Quiche(crate::quiche::QuicheRequest),
523 #[cfg(feature = "iroh")]
524 Iroh(crate::iroh::IrohRequest),
525 #[cfg(feature = "websocket")]
526 WebSocket(qmux::Session),
527}
528
529pub struct Request {
534 server: moq_lite::Server,
535 kind: RequestKind,
536}
537
538impl Request {
539 pub async fn close(self, _code: u16) -> anyhow::Result<()> {
541 match self.kind {
542 #[cfg(feature = "noq")]
543 RequestKind::Noq(request) => {
544 let status = web_transport_noq::http::StatusCode::from_u16(_code).context("invalid status code")?;
545 request.close(status).await?;
546 Ok(())
547 }
548 #[cfg(feature = "quinn")]
549 RequestKind::Quinn(request) => {
550 let status = web_transport_quinn::http::StatusCode::from_u16(_code).context("invalid status code")?;
551 request.close(status).await?;
552 Ok(())
553 }
554 #[cfg(feature = "quiche")]
555 RequestKind::Quiche(request) => {
556 let status = web_transport_quiche::http::StatusCode::from_u16(_code).context("invalid status code")?;
557 request
558 .reject(status)
559 .await
560 .map_err(|e| anyhow::anyhow!("failed to close quiche WebTransport request: {e}"))?;
561 Ok(())
562 }
563 #[cfg(feature = "iroh")]
564 RequestKind::Iroh(request) => {
565 let status = web_transport_iroh::http::StatusCode::from_u16(_code).context("invalid status code")?;
566 request.close(status).await?;
567 Ok(())
568 }
569 #[cfg(feature = "websocket")]
570 RequestKind::WebSocket(_session) => {
571 Ok(())
573 }
574 }
575 }
576
577 pub fn with_publish(mut self, publish: impl Into<Option<moq_lite::OriginConsumer>>) -> Self {
579 self.server = self.server.with_publish(publish);
580 self
581 }
582
583 pub fn with_consume(mut self, consume: impl Into<Option<moq_lite::OriginProducer>>) -> Self {
585 self.server = self.server.with_consume(consume);
586 self
587 }
588
589 pub async fn ok(self) -> anyhow::Result<Session> {
591 match self.kind {
592 #[cfg(feature = "noq")]
593 RequestKind::Noq(request) => Ok(self.server.accept(request.ok().await?).await?),
594 #[cfg(feature = "quinn")]
595 RequestKind::Quinn(request) => Ok(self.server.accept(request.ok().await?).await?),
596 #[cfg(feature = "quiche")]
597 RequestKind::Quiche(request) => {
598 let conn = request
599 .ok()
600 .await
601 .map_err(|e| anyhow::anyhow!("failed to accept quiche WebTransport: {e}"))?;
602 Ok(self.server.accept(conn).await?)
603 }
604 #[cfg(feature = "iroh")]
605 RequestKind::Iroh(request) => Ok(self.server.accept(request.ok().await?).await?),
606 #[cfg(feature = "websocket")]
607 RequestKind::WebSocket(session) => Ok(self.server.accept(session).await?),
608 }
609 }
610
611 pub fn transport(&self) -> &'static str {
613 match self.kind {
614 #[cfg(feature = "noq")]
615 RequestKind::Noq(_) => "quic",
616 #[cfg(feature = "quinn")]
617 RequestKind::Quinn(_) => "quic",
618 #[cfg(feature = "quiche")]
619 RequestKind::Quiche(_) => "quic",
620 #[cfg(feature = "iroh")]
621 RequestKind::Iroh(_) => "iroh",
622 #[cfg(feature = "websocket")]
623 RequestKind::WebSocket(_) => "websocket",
624 }
625 }
626
627 pub fn url(&self) -> Option<&Url> {
629 #[cfg(not(any(feature = "noq", feature = "quinn", feature = "quiche", feature = "iroh")))]
630 unreachable!("no QUIC backend compiled; enable noq, quinn, quiche, or iroh feature");
631
632 match self.kind {
633 #[cfg(feature = "noq")]
634 RequestKind::Noq(ref request) => request.url(),
635 #[cfg(feature = "quinn")]
636 RequestKind::Quinn(ref request) => request.url(),
637 #[cfg(feature = "quiche")]
638 RequestKind::Quiche(ref request) => request.url(),
639 #[cfg(feature = "iroh")]
640 RequestKind::Iroh(ref request) => request.url(),
641 #[cfg(feature = "websocket")]
642 RequestKind::WebSocket(_) => None,
643 }
644 }
645
646 pub fn peer_identity(&self) -> anyhow::Result<Option<PeerIdentity>> {
652 match self.kind {
653 #[cfg(feature = "quinn")]
654 RequestKind::Quinn(ref request) => request.peer_identity(),
655 #[cfg(feature = "noq")]
656 RequestKind::Noq(_) => Ok(None),
657 #[cfg(feature = "quiche")]
658 RequestKind::Quiche(_) => Ok(None),
659 #[cfg(feature = "iroh")]
660 RequestKind::Iroh(_) => Ok(None),
661 #[cfg(feature = "websocket")]
662 RequestKind::WebSocket(_) => Ok(None),
663 #[cfg(not(any(
664 feature = "noq",
665 feature = "quinn",
666 feature = "quiche",
667 feature = "iroh",
668 feature = "websocket"
669 )))]
670 _ => Ok(None),
671 }
672 }
673}
674
675#[derive(Debug)]
677pub struct ServerTlsInfo {
678 #[cfg(any(feature = "noq", feature = "quinn"))]
679 pub(crate) certs: Vec<Arc<rustls::sign::CertifiedKey>>,
680 pub fingerprints: Vec<String>,
681}
682
683#[serde_with::serde_as]
685#[derive(Clone, serde::Serialize, serde::Deserialize)]
686pub struct ServerId(#[serde_as(as = "serde_with::hex::Hex")] pub(crate) Vec<u8>);
687
688impl ServerId {
689 #[allow(dead_code)]
690 pub(crate) fn len(&self) -> usize {
691 self.0.len()
692 }
693}
694
695impl std::fmt::Debug for ServerId {
696 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
697 f.debug_tuple("QuicLbServerId").field(&hex::encode(&self.0)).finish()
698 }
699}
700
701impl std::str::FromStr for ServerId {
702 type Err = hex::FromHexError;
703
704 fn from_str(s: &str) -> Result<Self, Self::Err> {
705 hex::decode(s).map(Self)
706 }
707}
708
709#[cfg(test)]
710mod tests {
711 use super::*;
712
713 #[test]
714 fn test_tls_string_or_array() {
715 let single = r#"
717 cert = "cert.pem"
718 key = "key.pem"
719 "#;
720 let config: ServerTlsConfig = toml::from_str(single).unwrap();
721 assert_eq!(config.cert, vec![PathBuf::from("cert.pem")]);
722 assert_eq!(config.key, vec![PathBuf::from("key.pem")]);
723
724 let array = r#"
726 cert = ["a.pem", "b.pem"]
727 key = ["a.key", "b.key"]
728 generate = ["localhost"]
729 root = ["ca.pem"]
730 "#;
731 let config: ServerTlsConfig = toml::from_str(array).unwrap();
732 assert_eq!(config.cert, vec![PathBuf::from("a.pem"), PathBuf::from("b.pem")]);
733 assert_eq!(config.key, vec![PathBuf::from("a.key"), PathBuf::from("b.key")]);
734 assert_eq!(config.generate, vec!["localhost".to_string()]);
735 assert_eq!(config.root, vec![PathBuf::from("ca.pem")]);
736 }
737}