ferron/
server.rs

1use std::error::Error;
2use std::net::{IpAddr, Ipv6Addr, SocketAddr};
3use std::sync::Arc;
4use std::{env, thread};
5
6use crate::ferron_request_handler::request_handler;
7use crate::ferron_util::load_tls::{load_certs, load_private_key};
8use crate::ferron_util::sni::CustomSniResolver;
9use crate::ferron_util::validate_config::{prepare_config_for_validation, validate_config};
10
11use crate::ferron_common::{LogMessage, ServerModule, ServerModuleHandlers};
12use async_channel::Sender;
13use chrono::prelude::*;
14use futures_util::StreamExt;
15use h3_quinn::quinn;
16use h3_quinn::quinn::crypto::rustls::QuicServerConfig;
17use http::Response;
18use http_body_util::{BodyExt, StreamBody};
19use hyper::body::{Buf, Bytes, Frame, Incoming};
20use hyper::service::service_fn;
21use hyper::Request;
22use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
23use ocsp_stapler::Stapler;
24use rustls::crypto::ring::cipher_suite::*;
25use rustls::crypto::ring::default_provider;
26use rustls::crypto::ring::kx_group::*;
27use rustls::server::{Acceptor, WebPkiClientVerifier};
28use rustls::sign::CertifiedKey;
29use rustls::version::{TLS12, TLS13};
30use rustls::{RootCertStore, ServerConfig};
31use rustls_acme::acme::ACME_TLS_ALPN_NAME;
32use rustls_acme::caches::DirCache;
33use rustls_acme::{is_tls_alpn_challenge, AcmeConfig, ResolvesServerCertAcme, UseChallenge};
34use rustls_native_certs::load_native_certs;
35use tokio::fs;
36use tokio::io::{AsyncWriteExt, BufWriter};
37use tokio::net::{TcpListener, TcpStream};
38use tokio::runtime::Handle;
39use tokio::sync::Mutex;
40use tokio::time;
41use tokio_rustls::server::TlsStream;
42use tokio_rustls::LazyConfigAcceptor;
43use yaml_rust2::Yaml;
44
45// Enum for maybe TLS stream
46enum MaybeTlsStream {
47  Tls(TlsStream<TcpStream>),
48  Plain(TcpStream),
49}
50
51// Function to accept and handle incoming QUIC connections
52#[allow(clippy::too_many_arguments)]
53async fn accept_quic_connection(
54  connection_attempt: quinn::Incoming,
55  local_address: SocketAddr,
56  config: Arc<Yaml>,
57  logger: Sender<LogMessage>,
58  modules: Arc<Vec<Box<dyn ServerModule + std::marker::Send + Sync>>>,
59) {
60  let remote_address = connection_attempt.remote_address();
61
62  let logger_clone = logger.clone();
63
64  tokio::task::spawn(async move {
65    match connection_attempt.await {
66      Ok(connection) => {
67        let mut h3_conn: h3::server::Connection<h3_quinn::Connection, Bytes> =
68          match h3::server::Connection::new(h3_quinn::Connection::new(connection)).await {
69            Ok(h3_conn) => h3_conn,
70            Err(err) => {
71              logger_clone
72                .send(LogMessage::new(
73                  format!("Error serving HTTP/3 connection: {}", err),
74                  true,
75                ))
76                .await
77                .unwrap_or_default();
78              return;
79            }
80          };
81
82        loop {
83          match h3_conn.accept().await {
84            Ok(Some(resolver)) => {
85              let config = config.clone();
86              let remote_address = remote_address;
87
88              let logger_clone = logger_clone.clone();
89              let modules = modules.clone();
90              tokio::spawn(async move {
91                let handlers_vec = modules
92                  .iter()
93                  .map(|module| module.get_handlers(Handle::current()));
94
95                let (request, stream) = resolver;
96                let (mut send, receive) = stream.split();
97                let request_body_stream = futures_util::stream::unfold(
98                  (receive, false),
99                  async move |(mut receive, mut is_body_finished)| loop {
100                    if !is_body_finished {
101                      match receive.recv_data().await {
102                        Ok(Some(mut data)) => {
103                          return Some((
104                            Ok(Frame::data(data.copy_to_bytes(data.remaining()))),
105                            (receive, false),
106                          ))
107                        }
108                        Ok(None) => is_body_finished = true,
109                        Err(err) => {
110                          return Some((
111                            Err(std::io::Error::other(err.to_string())),
112                            (receive, false),
113                          ))
114                        }
115                      }
116                    } else {
117                      match receive.recv_trailers().await {
118                        Ok(Some(trailers)) => {
119                          return Some((Ok(Frame::trailers(trailers)), (receive, true)))
120                        }
121                        Ok(None) => is_body_finished = true,
122                        Err(err) => {
123                          return Some((
124                            Err(std::io::Error::other(err.to_string())),
125                            (receive, true),
126                          ))
127                        }
128                      }
129                    }
130                  },
131                );
132                let request_body = BodyExt::boxed(StreamBody::new(request_body_stream));
133                let (request_parts, _) = request.into_parts();
134                let request = Request::from_parts(request_parts, request_body);
135                let handlers_vec_clone = handlers_vec
136                  .clone()
137                  .collect::<Vec<Box<dyn ServerModuleHandlers + Send>>>();
138                let response = match request_handler(
139                  request,
140                  remote_address,
141                  local_address,
142                  true,
143                  config,
144                  logger_clone.clone(),
145                  handlers_vec_clone,
146                  None,
147                  None,
148                )
149                .await
150                {
151                  Ok(response) => response,
152                  Err(err) => {
153                    logger_clone
154                      .send(LogMessage::new(
155                        format!("Error serving HTTP/3 connection: {}", err),
156                        true,
157                      ))
158                      .await
159                      .unwrap_or_default();
160                    return;
161                  }
162                };
163                let (response_parts, mut response_body) = response.into_parts();
164                if let Err(err) = send
165                  .send_response(Response::from_parts(response_parts, ()))
166                  .await
167                {
168                  logger_clone
169                    .send(LogMessage::new(
170                      format!("Error serving HTTP/3 connection: {}", err),
171                      true,
172                    ))
173                    .await
174                    .unwrap_or_default();
175                  return;
176                }
177                let mut had_trailers = false;
178                while let Some(chunk) = response_body.frame().await {
179                  match chunk {
180                    Ok(frame) => {
181                      if frame.is_data() {
182                        match frame.into_data() {
183                          Ok(data) => {
184                            if let Err(err) = send.send_data(data).await {
185                              logger_clone
186                                .send(LogMessage::new(
187                                  format!("Error serving HTTP/3 connection: {}", err),
188                                  true,
189                                ))
190                                .await
191                                .unwrap_or_default();
192                              return;
193                            }
194                          }
195                          Err(_) => {
196                            logger_clone
197                            .send(LogMessage::new(
198                              "Error serving HTTP/3 connection: the frame isn't really a data frame".to_string(),
199                              true,
200                            ))
201                            .await
202                            .unwrap_or_default();
203                            return;
204                          }
205                        }
206                      } else if frame.is_trailers() {
207                        match frame.into_trailers() {
208                          Ok(trailers) => {
209                            had_trailers = true;
210                            if let Err(err) = send.send_trailers(trailers).await {
211                              logger_clone
212                                .send(LogMessage::new(
213                                  format!("Error serving HTTP/3 connection: {}", err),
214                                  true,
215                                ))
216                                .await
217                                .unwrap_or_default();
218                              return;
219                            }
220                          }
221                          Err(_) => {
222                            logger_clone
223                            .send(LogMessage::new(
224                              "Error serving HTTP/3 connection: the frame isn't really a trailers frame".to_string(),
225                              true,
226                            ))
227                            .await
228                            .unwrap_or_default();
229                            return;
230                          }
231                        }
232                      }
233                    }
234                    Err(err) => {
235                      logger_clone
236                        .send(LogMessage::new(
237                          format!("Error serving HTTP/3 connection: {}", err),
238                          true,
239                        ))
240                        .await
241                        .unwrap_or_default();
242                      return;
243                    }
244                  }
245                }
246                if !had_trailers {
247                  if let Err(err) = send.finish().await {
248                    logger_clone
249                      .send(LogMessage::new(
250                        format!("Error serving HTTP/3 connection: {}", err),
251                        true,
252                      ))
253                      .await
254                      .unwrap_or_default();
255                  }
256                }
257              });
258            }
259            Ok(None) => break,
260            Err(err) => {
261              logger_clone
262                .send(LogMessage::new(
263                  format!("Error serving HTTP/3 connection: {}", err),
264                  true,
265                ))
266                .await
267                .unwrap_or_default();
268              return;
269            }
270          }
271        }
272      }
273      Err(err) => {
274        logger_clone
275          .send(LogMessage::new(
276            format!("Cannot accept a connection: {}", err),
277            true,
278          ))
279          .await
280          .unwrap_or_default();
281      }
282    }
283  });
284}
285
286// Function to accept and handle incoming connections
287#[allow(clippy::too_many_arguments)]
288async fn accept_connection(
289  stream: TcpStream,
290  remote_address: SocketAddr,
291  tls_config_option: Option<(Arc<ServerConfig>, Option<Arc<ServerConfig>>)>,
292  acme_http01_resolver_option: Option<Arc<ResolvesServerCertAcme>>,
293  config: Arc<Yaml>,
294  logger: Sender<LogMessage>,
295  modules: Arc<Vec<Box<dyn ServerModule + std::marker::Send + Sync>>>,
296  http3_enabled: Option<u16>,
297) {
298  // Disable Nagle algorithm to improve performance
299  if let Err(err) = stream.set_nodelay(true) {
300    logger
301      .send(LogMessage::new(
302        format!("Cannot disable Nagle algorithm: {}", err),
303        true,
304      ))
305      .await
306      .unwrap_or_default();
307    return;
308  };
309
310  let config = config.clone();
311  let local_address = match stream.local_addr() {
312    Ok(local_address) => local_address,
313    Err(err) => {
314      logger
315        .send(LogMessage::new(
316          format!("Cannot obtain local address of the connection: {}", err),
317          true,
318        ))
319        .await
320        .unwrap_or_default();
321      return;
322    }
323  };
324
325  let logger_clone = logger.clone();
326
327  tokio::task::spawn(async move {
328    let maybe_tls_stream = if let Some((tls_config, acme_config_option)) = tls_config_option {
329      let start_handshake = match LazyConfigAcceptor::new(Acceptor::default(), stream).await {
330        Ok(start_handshake) => start_handshake,
331        Err(err) => {
332          logger
333            .send(LogMessage::new(
334              format!("Error during TLS handshake: {}", err),
335              true,
336            ))
337            .await
338            .unwrap_or_default();
339          return;
340        }
341      };
342
343      if let Some(acme_config) = acme_config_option {
344        if is_tls_alpn_challenge(&start_handshake.client_hello()) {
345          match start_handshake.into_stream(acme_config).await {
346            Ok(_) => (),
347            Err(err) => {
348              logger
349                .send(LogMessage::new(
350                  format!("Error during TLS handshake: {}", err),
351                  true,
352                ))
353                .await
354                .unwrap_or_default();
355              return;
356            }
357          };
358          return;
359        }
360      }
361
362      let tls_stream = match start_handshake.into_stream(tls_config).await {
363        Ok(tls_stream) => tls_stream,
364        Err(err) => {
365          logger
366            .send(LogMessage::new(
367              format!("Error during TLS handshake: {}", err),
368              true,
369            ))
370            .await
371            .unwrap_or_default();
372          return;
373        }
374      };
375
376      MaybeTlsStream::Tls(tls_stream)
377    } else {
378      MaybeTlsStream::Plain(stream)
379    };
380
381    if let MaybeTlsStream::Tls(tls_stream) = maybe_tls_stream {
382      let alpn_protocol = tls_stream.get_ref().1.alpn_protocol();
383      let is_http2;
384
385      if config["global"]["enableHTTP2"].as_bool().unwrap_or(true) {
386        if alpn_protocol == Some("h2".as_bytes()) {
387          is_http2 = true;
388        } else {
389          // Don't allow HTTP/2 if "h2" ALPN offering was't present
390          is_http2 = false;
391        }
392      } else {
393        is_http2 = false;
394      }
395
396      let io = TokioIo::new(tls_stream);
397      let handlers_vec = modules
398        .iter()
399        .map(|module| module.get_handlers(Handle::current()));
400
401      if is_http2 {
402        let mut http2_builder = hyper::server::conn::http2::Builder::new(TokioExecutor::new());
403        http2_builder.timer(TokioTimer::new());
404        if let Some(initial_window_size) =
405          config["global"]["http2Settings"]["initialWindowSize"].as_i64()
406        {
407          http2_builder.initial_stream_window_size(initial_window_size as u32);
408        }
409        if let Some(max_frame_size) = config["global"]["http2Settings"]["maxFrameSize"].as_i64() {
410          http2_builder.max_frame_size(max_frame_size as u32);
411        }
412        if let Some(max_concurrent_streams) =
413          config["global"]["http2Settings"]["maxConcurrentStreams"].as_i64()
414        {
415          http2_builder.max_concurrent_streams(max_concurrent_streams as u32);
416        }
417        if let Some(max_header_list_size) =
418          config["global"]["http2Settings"]["maxHeaderListSize"].as_i64()
419        {
420          http2_builder.max_header_list_size(max_header_list_size as u32);
421        }
422        if let Some(enable_connect_protocol) =
423          config["global"]["http2Settings"]["enableConnectProtocol"].as_bool()
424        {
425          if enable_connect_protocol {
426            http2_builder.enable_connect_protocol();
427          }
428        }
429
430        if let Err(err) = http2_builder
431          .serve_connection(
432            io,
433            service_fn(move |request: Request<Incoming>| {
434              let config = config.clone();
435              let logger = logger_clone.clone();
436              let handlers_vec_clone = handlers_vec
437                .clone()
438                .collect::<Vec<Box<dyn ServerModuleHandlers + Send>>>();
439              let acme_http01_resolver_option_clone = acme_http01_resolver_option.clone();
440              let (request_parts, request_body) = request.into_parts();
441              let request = Request::from_parts(
442                request_parts,
443                request_body
444                  .map_err(|e| std::io::Error::other(e.to_string()))
445                  .boxed(),
446              );
447              request_handler(
448                request,
449                remote_address,
450                local_address,
451                true,
452                config,
453                logger,
454                handlers_vec_clone,
455                acme_http01_resolver_option_clone,
456                http3_enabled,
457              )
458            }),
459          )
460          .await
461        {
462          logger
463            .send(LogMessage::new(
464              format!("Error serving HTTPS connection: {}", err),
465              true,
466            ))
467            .await
468            .unwrap_or_default();
469        }
470      } else {
471        let mut http1_builder = hyper::server::conn::http1::Builder::new();
472
473        // The timer is neccessary for the header timeout to work to mitigate Slowloris.
474        http1_builder.timer(TokioTimer::new());
475
476        if let Err(err) = http1_builder
477          .serve_connection(
478            io,
479            service_fn(move |request: Request<Incoming>| {
480              let config = config.clone();
481              let logger = logger_clone.clone();
482              let handlers_vec_clone = handlers_vec
483                .clone()
484                .collect::<Vec<Box<dyn ServerModuleHandlers + Send>>>();
485              let acme_http01_resolver_option_clone = acme_http01_resolver_option.clone();
486              let (request_parts, request_body) = request.into_parts();
487              let request = Request::from_parts(
488                request_parts,
489                request_body
490                  .map_err(|e| std::io::Error::other(e.to_string()))
491                  .boxed(),
492              );
493              request_handler(
494                request,
495                remote_address,
496                local_address,
497                true,
498                config,
499                logger,
500                handlers_vec_clone,
501                acme_http01_resolver_option_clone,
502                http3_enabled,
503              )
504            }),
505          )
506          .with_upgrades()
507          .await
508        {
509          logger
510            .send(LogMessage::new(
511              format!("Error serving HTTPS connection: {}", err),
512              true,
513            ))
514            .await
515            .unwrap_or_default();
516        }
517      }
518    } else if let MaybeTlsStream::Plain(stream) = maybe_tls_stream {
519      let io = TokioIo::new(stream);
520      let handlers_vec = modules
521        .iter()
522        .map(|module| module.get_handlers(Handle::current()));
523
524      let mut http1_builder = hyper::server::conn::http1::Builder::new();
525
526      // The timer is neccessary for the header timeout to work to mitigate Slowloris.
527      http1_builder.timer(TokioTimer::new());
528
529      if let Err(err) = http1_builder
530        .serve_connection(
531          io,
532          service_fn(move |request: Request<Incoming>| {
533            let config = config.clone();
534            let logger = logger_clone.clone();
535            let handlers_vec_clone = handlers_vec
536              .clone()
537              .collect::<Vec<Box<dyn ServerModuleHandlers + Send>>>();
538            let acme_http01_resolver_option_clone = acme_http01_resolver_option.clone();
539            let (request_parts, request_body) = request.into_parts();
540            let request = Request::from_parts(
541              request_parts,
542              request_body
543                .map_err(|e| std::io::Error::other(e.to_string()))
544                .boxed(),
545            );
546            request_handler(
547              request,
548              remote_address,
549              local_address,
550              false,
551              config,
552              logger,
553              handlers_vec_clone,
554              acme_http01_resolver_option_clone,
555              http3_enabled,
556            )
557          }),
558        )
559        .with_upgrades()
560        .await
561      {
562        logger
563          .send(LogMessage::new(
564            format!("Error serving HTTP connection: {}", err),
565            true,
566          ))
567          .await
568          .unwrap_or_default();
569      }
570    }
571  });
572}
573
574// Main server event loop
575#[allow(clippy::type_complexity)]
576async fn server_event_loop(
577  yaml_config: Arc<Yaml>,
578  logger: Sender<LogMessage>,
579  modules: Vec<Box<dyn ServerModule + Send + Sync>>,
580  module_error: Option<anyhow::Error>,
581  modules_optional_builtin: Vec<String>,
582  first_startup: bool,
583) -> Result<(), Box<dyn Error + Send + Sync>> {
584  if let Some(module_error) = module_error {
585    logger
586      .send(LogMessage::new(module_error.to_string(), true))
587      .await
588      .unwrap_or_default();
589    Err(module_error)?
590  }
591
592  let prepared_config = match prepare_config_for_validation(&yaml_config) {
593    Ok(prepared_config) => prepared_config,
594    Err(err) => {
595      logger
596        .send(LogMessage::new(
597          format!("Server configuration validation failed: {}", err),
598          true,
599        ))
600        .await
601        .unwrap_or_default();
602      Err(anyhow::anyhow!(format!(
603        "Server configuration validation failed: {}",
604        err
605      )))?
606    }
607  };
608
609  for (config_to_validate, is_global, is_location) in prepared_config {
610    match validate_config(
611      config_to_validate,
612      is_global,
613      is_location,
614      &modules_optional_builtin,
615    ) {
616      Ok(unused_properties) => {
617        for unused_property in unused_properties {
618          logger
619            .send(LogMessage::new(
620              format!(
621                "Unused configuration property detected: \"{}\". You might load an appropriate module to use this configuration property",
622                unused_property
623              ),
624              true,
625            ))
626            .await
627            .unwrap_or_default();
628        }
629      }
630      Err(err) => {
631        logger
632          .send(LogMessage::new(
633            format!("Server configuration validation failed: {}", err),
634            true,
635          ))
636          .await
637          .unwrap_or_default();
638        Err(anyhow::anyhow!(format!(
639          "Server configuration validation failed: {}",
640          err
641        )))?
642      }
643    };
644  }
645
646  let mut crypto_provider = default_provider();
647
648  if let Some(cipher_suite) = yaml_config["global"]["cipherSuite"].as_vec() {
649    let mut cipher_suites = Vec::new();
650    let cipher_suite_iter = cipher_suite.iter();
651    for cipher_suite_yaml in cipher_suite_iter {
652      if let Some(cipher_suite) = cipher_suite_yaml.as_str() {
653        let cipher_suite_to_add = match cipher_suite {
654          "TLS_AES_128_GCM_SHA256" => TLS13_AES_128_GCM_SHA256,
655          "TLS_AES_256_GCM_SHA384" => TLS13_AES_256_GCM_SHA384,
656          "TLS_CHACHA20_POLY1305_SHA256" => TLS13_CHACHA20_POLY1305_SHA256,
657          "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256" => TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
658          "TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384" => TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
659          "TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256" => {
660            TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256
661          }
662          "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256" => TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
663          "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" => TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
664          "TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256" => {
665            TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256
666          }
667          _ => {
668            logger
669              .send(LogMessage::new(
670                format!("The \"{}\" cipher suite is not supported", cipher_suite),
671                true,
672              ))
673              .await
674              .unwrap_or_default();
675            Err(anyhow::anyhow!(format!(
676              "The \"{}\" cipher suite is not supported",
677              cipher_suite
678            )))?
679          }
680        };
681        cipher_suites.push(cipher_suite_to_add);
682      }
683    }
684    crypto_provider.cipher_suites = cipher_suites;
685  }
686
687  if let Some(ecdh_curves) = yaml_config["global"]["ecdhCurve"].as_vec() {
688    let mut kx_groups = Vec::new();
689    let ecdh_curves_iter = ecdh_curves.iter();
690    for ecdh_curve_yaml in ecdh_curves_iter {
691      if let Some(ecdh_curve) = ecdh_curve_yaml.as_str() {
692        let kx_group_to_add = match ecdh_curve {
693          "secp256r1" => SECP256R1,
694          "secp384r1" => SECP384R1,
695          "x25519" => X25519,
696          _ => {
697            logger
698              .send(LogMessage::new(
699                format!("The \"{}\" ECDH curve is not supported", ecdh_curve),
700                true,
701              ))
702              .await
703              .unwrap_or_default();
704            Err(anyhow::anyhow!(format!(
705              "The \"{}\" ECDH curve is not supported",
706              ecdh_curve
707            )))?
708          }
709        };
710        kx_groups.push(kx_group_to_add);
711      }
712    }
713    crypto_provider.kx_groups = kx_groups;
714  }
715
716  let crypto_provider_cloned = crypto_provider.clone();
717  let mut sni_resolver = CustomSniResolver::new();
718  let mut certified_keys = Vec::new();
719
720  let mut automatic_tls_enabled = false;
721  let mut acme_letsencrypt_production = true;
722  let acme_use_http_challenge = yaml_config["global"]["useAutomaticTLSHTTPChallenge"]
723    .as_bool()
724    .unwrap_or(false);
725  let acme_challenge_type = if acme_use_http_challenge {
726    UseChallenge::Http01
727  } else {
728    UseChallenge::TlsAlpn01
729  };
730
731  // Read automatic TLS configuration
732  if let Some(read_automatic_tls_enabled) = yaml_config["global"]["enableAutomaticTLS"].as_bool() {
733    automatic_tls_enabled = read_automatic_tls_enabled;
734  }
735
736  let acme_contact = yaml_config["global"]["automaticTLSContactEmail"].as_str();
737  let acme_cache = yaml_config["global"]["automaticTLSContactCacheDirectory"]
738    .as_str()
739    .map(|s| s.to_string())
740    .map(DirCache::new);
741
742  if let Some(read_acme_letsencrypt_production) =
743    yaml_config["global"]["automaticTLSLetsEncryptProduction"].as_bool()
744  {
745    acme_letsencrypt_production = read_acme_letsencrypt_production;
746  }
747
748  if !automatic_tls_enabled {
749    // Load public certificate and private key
750    if let Some(cert_path) = yaml_config["global"]["cert"].as_str() {
751      if let Some(key_path) = yaml_config["global"]["key"].as_str() {
752        let certs = match load_certs(cert_path) {
753          Ok(certs) => certs,
754          Err(err) => {
755            logger
756              .send(LogMessage::new(
757                format!("Cannot load the \"{}\" TLS certificate: {}", cert_path, err),
758                true,
759              ))
760              .await
761              .unwrap_or_default();
762            Err(anyhow::anyhow!(format!(
763              "Cannot load the \"{}\" TLS certificate: {}",
764              cert_path, err
765            )))?
766          }
767        };
768        let key = match load_private_key(key_path) {
769          Ok(key) => key,
770          Err(err) => {
771            logger
772              .send(LogMessage::new(
773                format!("Cannot load the \"{}\" private key: {}", cert_path, err),
774                true,
775              ))
776              .await
777              .unwrap_or_default();
778            Err(anyhow::anyhow!(format!(
779              "Cannot load the \"{}\" private key: {}",
780              cert_path, err
781            )))?
782          }
783        };
784        let signing_key = match crypto_provider_cloned.key_provider.load_private_key(key) {
785          Ok(key) => key,
786          Err(err) => {
787            logger
788              .send(LogMessage::new(
789                format!("Cannot load the \"{}\" private key: {}", cert_path, err),
790                true,
791              ))
792              .await
793              .unwrap_or_default();
794            Err(anyhow::anyhow!(format!(
795              "Cannot load the \"{}\" private key: {}",
796              cert_path, err
797            )))?
798          }
799        };
800        let certified_key = CertifiedKey::new(certs, signing_key);
801        sni_resolver.load_fallback_cert_key(Arc::new(certified_key));
802      }
803    }
804
805    if let Some(sni) = yaml_config["global"]["sni"].as_hash() {
806      let sni_hostnames = sni.keys();
807      for sni_hostname_unknown in sni_hostnames {
808        if let Some(sni_hostname) = sni_hostname_unknown.as_str() {
809          if let Some(cert_path) = sni[sni_hostname_unknown]["cert"].as_str() {
810            if let Some(key_path) = sni[sni_hostname_unknown]["key"].as_str() {
811              let certs = match load_certs(cert_path) {
812                Ok(certs) => certs,
813                Err(err) => {
814                  logger
815                    .send(LogMessage::new(
816                      format!("Cannot load the \"{}\" TLS certificate: {}", cert_path, err),
817                      true,
818                    ))
819                    .await
820                    .unwrap_or_default();
821                  Err(anyhow::anyhow!(format!(
822                    "Cannot load the \"{}\" TLS certificate: {}",
823                    cert_path, err
824                  )))?
825                }
826              };
827              let key = match load_private_key(key_path) {
828                Ok(key) => key,
829                Err(err) => {
830                  logger
831                    .send(LogMessage::new(
832                      format!("Cannot load the \"{}\" private key: {}", cert_path, err),
833                      true,
834                    ))
835                    .await
836                    .unwrap_or_default();
837                  Err(anyhow::anyhow!(format!(
838                    "Cannot load the \"{}\" private key: {}",
839                    cert_path, err
840                  )))?
841                }
842              };
843              let signing_key = match crypto_provider_cloned.key_provider.load_private_key(key) {
844                Ok(key) => key,
845                Err(err) => {
846                  logger
847                    .send(LogMessage::new(
848                      format!("Cannot load the \"{}\" private key: {}", cert_path, err),
849                      true,
850                    ))
851                    .await
852                    .unwrap_or_default();
853                  Err(anyhow::anyhow!(format!(
854                    "Cannot load the \"{}\" private key: {}",
855                    cert_path, err
856                  )))?
857                }
858              };
859              let certified_key_arc = Arc::new(CertifiedKey::new(certs, signing_key));
860              sni_resolver.load_host_cert_key(sni_hostname, certified_key_arc.clone());
861              certified_keys.push(certified_key_arc);
862            }
863          }
864        }
865      }
866    }
867  }
868
869  // Build TLS configuration
870  let tls_config_builder_wants_versions =
871    ServerConfig::builder_with_provider(Arc::new(crypto_provider_cloned));
872
873  // Very simple minimum and maximum TLS version logic for now...
874  let min_tls_version_option = yaml_config["global"]["tlsMinVersion"].as_str();
875  let max_tls_version_option = yaml_config["global"]["tlsMaxVersion"].as_str();
876  let tls_config_builder_wants_verifier = match min_tls_version_option {
877    Some("TLSv1.3") => match max_tls_version_option {
878      Some("TLSv1.2") => {
879        logger
880          .send(LogMessage::new(
881            String::from("The maximum TLS version is older than the minimum TLS version"),
882            true,
883          ))
884          .await
885          .unwrap_or_default();
886        Err(anyhow::anyhow!(String::from(
887          "The maximum TLS version is older than the minimum TLS version"
888        )))?
889      }
890      Some("TLSv1.3") | None => {
891        match tls_config_builder_wants_versions.with_protocol_versions(&[&TLS13]) {
892          Ok(builder) => builder,
893          Err(err) => {
894            logger
895              .send(LogMessage::new(
896                format!("Couldn't create the TLS server configuration: {}", err),
897                true,
898              ))
899              .await
900              .unwrap_or_default();
901            Err(anyhow::anyhow!(format!(
902              "Couldn't create the TLS server configuration: {}",
903              err
904            )))?
905          }
906        }
907      }
908      _ => {
909        logger
910          .send(LogMessage::new(
911            String::from("Invalid maximum TLS version"),
912            true,
913          ))
914          .await
915          .unwrap_or_default();
916        Err(anyhow::anyhow!(String::from("Invalid maximum TLS version")))?
917      }
918    },
919    Some("TLSv1.2") | None => match max_tls_version_option {
920      Some("TLSv1.2") => {
921        match tls_config_builder_wants_versions.with_protocol_versions(&[&TLS12]) {
922          Ok(builder) => builder,
923          Err(err) => {
924            logger
925              .send(LogMessage::new(
926                format!("Couldn't create the TLS server configuration: {}", err),
927                true,
928              ))
929              .await
930              .unwrap_or_default();
931            Err(anyhow::anyhow!(format!(
932              "Couldn't create the TLS server configuration: {}",
933              err
934            )))?
935          }
936        }
937      }
938      Some("TLSv1.3") | None => {
939        match tls_config_builder_wants_versions.with_protocol_versions(&[&TLS12, &TLS13]) {
940          Ok(builder) => builder,
941          Err(err) => {
942            logger
943              .send(LogMessage::new(
944                format!("Couldn't create the TLS server configuration: {}", err),
945                true,
946              ))
947              .await
948              .unwrap_or_default();
949            Err(anyhow::anyhow!(format!(
950              "Couldn't create the TLS server configuration: {}",
951              err
952            )))?
953          }
954        }
955      }
956      _ => {
957        logger
958          .send(LogMessage::new(
959            String::from("Invalid maximum TLS version"),
960            true,
961          ))
962          .await
963          .unwrap_or_default();
964        Err(anyhow::anyhow!(String::from("Invalid maximum TLS version")))?
965      }
966    },
967    _ => {
968      logger
969        .send(LogMessage::new(
970          String::from("Invalid minimum TLS version"),
971          true,
972        ))
973        .await
974        .unwrap_or_default();
975      Err(anyhow::anyhow!(String::from("Invalid minimum TLS version")))?
976    }
977  };
978
979  let tls_config_builder_wants_server_cert =
980    match yaml_config["global"]["useClientCertificate"].as_bool() {
981      Some(true) => {
982        let mut roots = RootCertStore::empty();
983        let certs_result = load_native_certs();
984        if !certs_result.errors.is_empty() {
985          logger
986            .send(LogMessage::new(
987              format!(
988                "Couldn't load the native certificate store: {}",
989                certs_result.errors[0]
990              ),
991              true,
992            ))
993            .await
994            .unwrap_or_default();
995          Err(anyhow::anyhow!(format!(
996            "Couldn't load the native certificate store: {}",
997            certs_result.errors[0]
998          )))?
999        }
1000        let certs = certs_result.certs;
1001
1002        for cert in certs {
1003          match roots.add(cert) {
1004            Ok(_) => (),
1005            Err(err) => {
1006              logger
1007                .send(LogMessage::new(
1008                  format!(
1009                    "Couldn't add a certificate to the certificate store: {}",
1010                    err
1011                  ),
1012                  true,
1013                ))
1014                .await
1015                .unwrap_or_default();
1016              Err(anyhow::anyhow!(format!(
1017                "Couldn't add a certificate to the certificate store: {}",
1018                err
1019              )))?
1020            }
1021          }
1022        }
1023        tls_config_builder_wants_verifier
1024          .with_client_cert_verifier(WebPkiClientVerifier::builder(Arc::new(roots)).build()?)
1025      }
1026      _ => tls_config_builder_wants_verifier.with_no_client_auth(),
1027    };
1028
1029  let mut tls_config;
1030
1031  let mut addr = SocketAddr::from((IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0)), 80));
1032  let mut addr_tls = SocketAddr::from((IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0)), 443));
1033  let mut tls_enabled = false;
1034  let mut non_tls_disabled = false;
1035
1036  // Install a process-wide cryptography provider. If it fails, then warn about it.
1037  if crypto_provider.install_default().is_err() && first_startup {
1038    logger
1039      .send(LogMessage::new(
1040        "Cannot install a process-wide cryptography provider".to_string(),
1041        true,
1042      ))
1043      .await
1044      .unwrap_or_default();
1045    Err(anyhow::anyhow!(
1046      "Cannot install a process-wide cryptography provider"
1047    ))?;
1048  }
1049
1050  // Read port configurations from YAML
1051  if let Some(read_port) = yaml_config["global"]["port"].as_i64() {
1052    addr = SocketAddr::from((
1053      IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0)),
1054      match read_port.try_into() {
1055        Ok(port) => port,
1056        Err(_) => {
1057          logger
1058            .send(LogMessage::new(String::from("Invalid HTTP port"), true))
1059            .await
1060            .unwrap_or_default();
1061          Err(anyhow::anyhow!("Invalid HTTP port"))?
1062        }
1063      },
1064    ));
1065  } else if let Some(read_port) = yaml_config["global"]["port"].as_str() {
1066    addr = match read_port.parse() {
1067      Ok(addr) => addr,
1068      Err(_) => {
1069        logger
1070          .send(LogMessage::new(String::from("Invalid HTTP port"), true))
1071          .await
1072          .unwrap_or_default();
1073        Err(anyhow::anyhow!("Invalid HTTP port"))?
1074      }
1075    };
1076  }
1077
1078  if let Some(read_tls_enabled) = yaml_config["global"]["secure"].as_bool() {
1079    tls_enabled = read_tls_enabled;
1080    if let Some(read_non_tls_disabled) =
1081      yaml_config["global"]["disableNonEncryptedServer"].as_bool()
1082    {
1083      non_tls_disabled = read_non_tls_disabled;
1084    }
1085  }
1086
1087  if let Some(read_port) = yaml_config["global"]["sport"].as_i64() {
1088    addr_tls = SocketAddr::from((
1089      IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0)),
1090      match read_port.try_into() {
1091        Ok(port) => port,
1092        Err(_) => {
1093          logger
1094            .send(LogMessage::new(String::from("Invalid HTTPS port"), true))
1095            .await
1096            .unwrap_or_default();
1097          Err(anyhow::anyhow!("Invalid HTTPS port"))?
1098        }
1099      },
1100    ));
1101  } else if let Some(read_port) = yaml_config["global"]["sport"].as_str() {
1102    addr_tls = match read_port.parse() {
1103      Ok(addr) => addr,
1104      Err(_) => {
1105        logger
1106          .send(LogMessage::new(String::from("Invalid HTTPS port"), true))
1107          .await
1108          .unwrap_or_default();
1109        Err(anyhow::anyhow!("Invalid HTTPS port"))?
1110      }
1111    };
1112  }
1113
1114  // Get domains for ACME configuration
1115  let mut acme_domains = Vec::new();
1116  if let Some(hosts_config) = yaml_config["hosts"].as_vec() {
1117    for host_yaml in hosts_config.iter() {
1118      if let Some(host) = host_yaml.as_hash() {
1119        if let Some(domain_yaml) = host.get(&Yaml::from_str("domain")) {
1120          if let Some(domain) = domain_yaml.as_str() {
1121            if !domain.contains("*") {
1122              acme_domains.push(domain);
1123            }
1124          }
1125        }
1126      }
1127    }
1128  }
1129
1130  // Create ACME configuration
1131  let mut acme_config = AcmeConfig::new(acme_domains).challenge_type(acme_challenge_type);
1132  if let Some(acme_contact_unwrapped) = acme_contact {
1133    acme_config = acme_config.contact_push(format!("mailto:{}", acme_contact_unwrapped));
1134  }
1135  let mut acme_config_with_cache = acme_config.cache_option(acme_cache);
1136  acme_config_with_cache =
1137    acme_config_with_cache.directory_lets_encrypt(acme_letsencrypt_production);
1138
1139  let (acme_config, acme_http01_resolver) = if tls_enabled && automatic_tls_enabled {
1140    let mut acme_state = acme_config_with_cache.state();
1141
1142    let acme_resolver = acme_state.resolver();
1143
1144    // Create TLS configuration
1145    tls_config = if yaml_config["global"]["enableOCSPStapling"]
1146      .as_bool()
1147      .unwrap_or(true)
1148    {
1149      tls_config_builder_wants_server_cert
1150        .with_cert_resolver(Arc::new(Stapler::new(acme_resolver.clone())))
1151    } else {
1152      tls_config_builder_wants_server_cert.with_cert_resolver(acme_resolver.clone())
1153    };
1154
1155    let acme_logger = logger.clone();
1156    tokio::spawn(async move {
1157      while let Some(acme_result) = acme_state.next().await {
1158        if let Err(acme_error) = acme_result {
1159          acme_logger
1160            .send(LogMessage::new(
1161              format!("Error while obtaining a TLS certificate: {}", acme_error),
1162              true,
1163            ))
1164            .await
1165            .unwrap_or_default();
1166        }
1167      }
1168    });
1169
1170    if acme_use_http_challenge {
1171      (None, Some(acme_resolver))
1172    } else {
1173      let mut acme_config = tls_config.clone();
1174      acme_config.alpn_protocols.push(ACME_TLS_ALPN_NAME.to_vec());
1175
1176      (Some(acme_config), None)
1177    }
1178  } else {
1179    // Create TLS configuration
1180    tls_config = if yaml_config["global"]["enableOCSPStapling"]
1181      .as_bool()
1182      .unwrap_or(true)
1183    {
1184      let ocsp_stapler_arc = Arc::new(Stapler::new(Arc::new(sni_resolver)));
1185      for certified_key in certified_keys.iter() {
1186        ocsp_stapler_arc.preload(certified_key.clone());
1187      }
1188      tls_config_builder_wants_server_cert.with_cert_resolver(ocsp_stapler_arc.clone())
1189    } else {
1190      tls_config_builder_wants_server_cert.with_cert_resolver(Arc::new(sni_resolver))
1191    };
1192
1193    // Drop the ACME configuration
1194    drop(acme_config_with_cache);
1195    (None, None)
1196  };
1197
1198  let quic_config = if tls_enabled
1199    && yaml_config["global"]["enableHTTP3"]
1200      .as_bool()
1201      .unwrap_or(false)
1202  {
1203    let mut quic_tls_config = tls_config.clone();
1204    quic_tls_config.max_early_data_size = u32::MAX;
1205    quic_tls_config.alpn_protocols = vec![b"h3".to_vec(), b"h3-29".to_vec()];
1206    let quic_config = quinn::ServerConfig::with_crypto(Arc::new(match QuicServerConfig::try_from(
1207      quic_tls_config,
1208    ) {
1209      Ok(quinn_config) => quinn_config,
1210      Err(err) => {
1211        logger
1212          .send(LogMessage::new(
1213            format!("There was a problem when starting HTTP/3 server: {}", err),
1214            true,
1215          ))
1216          .await
1217          .unwrap_or_default();
1218        Err(anyhow::anyhow!(format!(
1219          "There was a problem when starting HTTP/3 server: {}",
1220          err
1221        )))?
1222      }
1223    }));
1224    Some(quic_config)
1225  } else {
1226    None
1227  };
1228
1229  // Configure ALPN protocols
1230  let mut alpn_protocols = vec![b"http/1.1".to_vec(), b"http/1.0".to_vec()];
1231  if yaml_config["global"]["enableHTTP2"]
1232    .as_bool()
1233    .unwrap_or(true)
1234  {
1235    alpn_protocols.insert(0, b"h2".to_vec());
1236  }
1237  tls_config.alpn_protocols = alpn_protocols;
1238  let tls_config_arc = Arc::new(tls_config);
1239  let acme_config_arc = acme_config.map(Arc::new);
1240
1241  let mut listener = None;
1242  let mut listener_tls = None;
1243  let mut listener_quic = None;
1244
1245  // Bind to the specified ports
1246  if !non_tls_disabled {
1247    println!("HTTP server is listening at {}", addr);
1248    listener = Some(match TcpListener::bind(addr).await {
1249      Ok(listener) => listener,
1250      Err(err) => {
1251        logger
1252          .send(LogMessage::new(
1253            format!("Cannot listen to HTTP port: {}", err),
1254            true,
1255          ))
1256          .await
1257          .unwrap_or_default();
1258        Err(anyhow::anyhow!(format!(
1259          "Cannot listen to HTTP port: {}",
1260          err
1261        )))?
1262      }
1263    });
1264  }
1265
1266  if tls_enabled {
1267    println!("HTTPS server is listening at {}", addr_tls);
1268    listener_tls = Some(match TcpListener::bind(addr_tls).await {
1269      Ok(listener) => listener,
1270      Err(err) => {
1271        logger
1272          .send(LogMessage::new(
1273            format!("Cannot listen to HTTPS port: {}", err),
1274            true,
1275          ))
1276          .await
1277          .unwrap_or_default();
1278        Err(anyhow::anyhow!(format!(
1279          "Cannot listen to HTTPS port: {}",
1280          err
1281        )))?
1282      }
1283    });
1284
1285    if let Some(quic_config) = quic_config {
1286      println!("HTTP/3 server is listening at {}", addr_tls);
1287      listener_quic = Some(match quinn::Endpoint::server(quic_config, addr_tls) {
1288        Ok(listener) => listener,
1289        Err(err) => {
1290          logger
1291            .send(LogMessage::new(
1292              format!("Cannot listen to HTTP/3 port: {}", err),
1293              true,
1294            ))
1295            .await
1296            .unwrap_or_default();
1297          Err(anyhow::anyhow!(format!(
1298            "Cannot listen to HTTP/3 port: {}",
1299            err
1300          )))?
1301        }
1302      });
1303    }
1304  }
1305
1306  // Wrap the modules vector in an Arc
1307  let modules_arc = Arc::new(modules);
1308
1309  let http3_enabled = if listener_tls.is_some() {
1310    Some(addr_tls.port())
1311  } else {
1312    None
1313  };
1314
1315  // Main loop to accept incoming connections
1316  loop {
1317    let listener_borrowed = &listener;
1318    let listener_accept = async move {
1319      if let Some(listener) = listener_borrowed {
1320        listener.accept().await
1321      } else {
1322        futures_util::future::pending().await
1323      }
1324    };
1325
1326    let listener_tls_borrowed = &listener_tls;
1327    let listener_tls_accept = async move {
1328      if let Some(listener_tls) = listener_tls_borrowed {
1329        listener_tls.accept().await
1330      } else {
1331        futures_util::future::pending().await
1332      }
1333    };
1334
1335    let listener_quic_borrowed = &listener_quic;
1336    let listener_quic_accept = async move {
1337      if let Some(listener_quic) = listener_quic_borrowed {
1338        listener_quic.accept().await
1339      } else {
1340        futures_util::future::pending().await
1341      }
1342    };
1343
1344    if listener_borrowed.is_none()
1345      && listener_tls_borrowed.is_none()
1346      && listener_quic_borrowed.is_none()
1347    {
1348      logger
1349        .send(LogMessage::new(
1350          String::from("No server is listening"),
1351          true,
1352        ))
1353        .await
1354        .unwrap_or_default();
1355      Err(anyhow::anyhow!("No server is listening"))?;
1356    }
1357
1358    tokio::select! {
1359      status = listener_accept => {
1360        match status {
1361          Ok((stream, remote_address)) => {
1362            accept_connection(
1363              stream,
1364              remote_address,
1365              None,
1366              acme_http01_resolver.clone(),
1367              yaml_config.clone(),
1368              logger.clone(),
1369              modules_arc.clone(),
1370              None
1371            )
1372            .await;
1373          }
1374          Err(err) => {
1375            logger
1376              .send(LogMessage::new(
1377                format!("Cannot accept a connection: {}", err),
1378                true,
1379              ))
1380              .await
1381              .unwrap_or_default();
1382          }
1383        }
1384      },
1385      status = listener_tls_accept => {
1386        match status {
1387          Ok((stream, remote_address)) => {
1388            accept_connection(
1389              stream,
1390              remote_address,
1391              Some((tls_config_arc.clone(), acme_config_arc.clone())),
1392              None,
1393              yaml_config.clone(),
1394              logger.clone(),
1395              modules_arc.clone(),
1396              http3_enabled
1397            )
1398            .await;
1399          }
1400          Err(err) => {
1401            logger
1402              .send(LogMessage::new(
1403                format!("Cannot accept a connection: {}", err),
1404                true,
1405              ))
1406              .await
1407              .unwrap_or_default();
1408          }
1409        }
1410      },
1411      status = listener_quic_accept => {
1412        match status {
1413          Some(connection_attempt) => {
1414            let local_ip = SocketAddr::new(connection_attempt.local_ip().unwrap_or(IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0))), addr_tls.port());
1415            accept_quic_connection(
1416              connection_attempt,
1417              local_ip,
1418              yaml_config.clone(),
1419              logger.clone(),
1420              modules_arc.clone()
1421            )
1422            .await;
1423          }
1424          None => {
1425            logger
1426              .send(LogMessage::new(
1427                "HTTP/3 connections can't be accepted anymore".to_string(),
1428                true,
1429              ))
1430              .await
1431              .unwrap_or_default();
1432          }
1433        }
1434      }
1435    };
1436  }
1437}
1438
1439// Start the server
1440#[allow(clippy::type_complexity)]
1441pub fn start_server(
1442  yaml_config: Arc<Yaml>,
1443  modules: Vec<Box<dyn ServerModule + Send + Sync>>,
1444  module_error: Option<anyhow::Error>,
1445  modules_optional_builtin: Vec<String>,
1446  first_startup: bool,
1447) -> Result<bool, Box<dyn Error + Send + Sync>> {
1448  if let Some(environment_variables_hash) = yaml_config["global"]["environmentVariables"].as_hash()
1449  {
1450    let environment_variables_hash_iter = environment_variables_hash.iter();
1451    for (variable_name, variable_value) in environment_variables_hash_iter {
1452      if let Some(variable_name) = variable_name.as_str() {
1453        if let Some(variable_value) = variable_value.as_str() {
1454          if !variable_name.is_empty()
1455            && !variable_name.contains('\0')
1456            && !variable_name.contains('=')
1457            && !variable_value.contains('\0')
1458          {
1459            // Safety: the environment variables are set before threads are spawned
1460            // The `std::env::set_var` function is safe to use in single-threaded environments
1461            // In Rust 2024 edition, the `std::env::set_var` function would be `unsafe`.
1462            env::set_var(variable_name, variable_value);
1463          }
1464        }
1465      }
1466    }
1467  }
1468
1469  let available_parallelism = thread::available_parallelism()?.get();
1470
1471  // Create Tokio runtime for the server
1472  let server_runtime = tokio::runtime::Builder::new_multi_thread()
1473    .worker_threads(available_parallelism)
1474    .max_blocking_threads(1536)
1475    .event_interval(25)
1476    .thread_name("server-pool")
1477    .enable_all()
1478    .build()?;
1479
1480  // Create Tokio runtime for logging
1481  let log_runtime = tokio::runtime::Builder::new_multi_thread()
1482    .worker_threads(match available_parallelism / 2 {
1483      0 => 1,
1484      non_zero => non_zero,
1485    })
1486    .max_blocking_threads(768)
1487    .thread_name("log-pool")
1488    .enable_time()
1489    .build()?;
1490
1491  let (logger, receive_log) = async_channel::bounded::<LogMessage>(10000);
1492
1493  let log_filename = yaml_config["global"]["logFilePath"]
1494    .as_str()
1495    .map(String::from);
1496  let error_log_filename = yaml_config["global"]["errorLogFilePath"]
1497    .as_str()
1498    .map(String::from);
1499
1500  log_runtime.spawn(async move {
1501    let log_file = match log_filename {
1502      Some(log_filename) => Some(
1503        fs::OpenOptions::new()
1504          .append(true)
1505          .create(true)
1506          .open(log_filename)
1507          .await,
1508      ),
1509      None => None,
1510    };
1511
1512    let error_log_file = match error_log_filename {
1513      Some(error_log_filename) => Some(
1514        fs::OpenOptions::new()
1515          .append(true)
1516          .create(true)
1517          .open(error_log_filename)
1518          .await,
1519      ),
1520      None => None,
1521    };
1522
1523    let log_file_wrapped = match log_file {
1524      Some(Ok(file)) => Some(Arc::new(Mutex::new(BufWriter::with_capacity(131072, file)))),
1525      Some(Err(e)) => {
1526        eprintln!("Failed to open log file: {}", e);
1527        None
1528      }
1529      None => None,
1530    };
1531
1532    let error_log_file_wrapped = match error_log_file {
1533      Some(Ok(file)) => Some(Arc::new(Mutex::new(BufWriter::with_capacity(131072, file)))),
1534      Some(Err(e)) => {
1535        eprintln!("Failed to open error log file: {}", e);
1536        None
1537      }
1538      None => None,
1539    };
1540
1541    // The logs are written when the log message is received by the log event loop, and flushed every 100 ms, improving the server performance.
1542    let log_file_wrapped_cloned_for_sleep = log_file_wrapped.clone();
1543    let error_log_file_wrapped_cloned_for_sleep = error_log_file_wrapped.clone();
1544    tokio::task::spawn(async move {
1545      let mut interval = time::interval(time::Duration::from_millis(100));
1546      loop {
1547        interval.tick().await;
1548        if let Some(log_file_wrapped_cloned) = log_file_wrapped_cloned_for_sleep.clone() {
1549          let mut locked_file = log_file_wrapped_cloned.lock().await;
1550          locked_file.flush().await.unwrap_or_default();
1551        }
1552        if let Some(error_log_file_wrapped_cloned) = error_log_file_wrapped_cloned_for_sleep.clone()
1553        {
1554          let mut locked_file = error_log_file_wrapped_cloned.lock().await;
1555          locked_file.flush().await.unwrap_or_default();
1556        }
1557      }
1558    });
1559
1560    // Logging loop
1561    while let Ok(message) = receive_log.recv().await {
1562      let (mut message, is_error) = message.get_message();
1563      let log_file_wrapped_cloned = if !is_error {
1564        log_file_wrapped.clone()
1565      } else {
1566        error_log_file_wrapped.clone()
1567      };
1568
1569      if let Some(log_file_wrapped_cloned) = log_file_wrapped_cloned {
1570        tokio::task::spawn(async move {
1571          let mut locked_file = log_file_wrapped_cloned.lock().await;
1572          if is_error {
1573            let now: DateTime<Local> = Local::now();
1574            let formatted_time = now.format("%Y-%m-%d %H:%M:%S").to_string();
1575            message = format!("[{}]: {}", formatted_time, message);
1576          }
1577          message.push('\n');
1578          if let Err(e) = locked_file.write(message.as_bytes()).await {
1579            eprintln!("Failed to write to log file: {}", e);
1580          }
1581        });
1582      }
1583    }
1584  });
1585
1586  // Run the server event loop
1587  let result = server_runtime.block_on(async {
1588    let event_loop_future = server_event_loop(
1589      yaml_config,
1590      logger,
1591      modules,
1592      module_error,
1593      modules_optional_builtin,
1594      first_startup,
1595    );
1596
1597    #[cfg(unix)]
1598    {
1599      use tokio::signal;
1600
1601      match signal::unix::signal(signal::unix::SignalKind::hangup()) {
1602        Ok(mut signal) => {
1603          tokio::select! {
1604            result = event_loop_future => {
1605              // Sleep the Tokio runtime to ensure error logs are saved
1606              time::sleep(tokio::time::Duration::from_millis(100)).await;
1607
1608              result.map(|_| false)
1609            },
1610            _ = signal.recv() => Ok(true)
1611          }
1612        }
1613        Err(_) => {
1614          let result = event_loop_future.await;
1615
1616          // Sleep the Tokio runtime to ensure error logs are saved
1617          time::sleep(tokio::time::Duration::from_millis(100)).await;
1618
1619          result.map(|_| false)
1620        }
1621      }
1622    }
1623
1624    #[cfg(not(unix))]
1625    {
1626      let result = event_loop_future.await;
1627
1628      // Sleep the Tokio runtime to ensure error logs are saved
1629      time::sleep(tokio::time::Duration::from_millis(100)).await;
1630
1631      result.map(|_| false)
1632    }
1633  });
1634
1635  // Wait 10 seconds or until all tasks are complete
1636  server_runtime.shutdown_timeout(time::Duration::from_secs(10));
1637
1638  result
1639}
ferron/server.rs

ferron/
server.rs
