ferron/

request_handler.rs

use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;

use async_channel::Sender;
use chrono::{DateTime, Local};
use ferron_common::logging::{ErrorLogger, LogMessage};
use ferron_common::observability::{MetricsMultiSender, TraceSignal};
use futures_util::stream::TryStreamExt;
use http_body_util::combinators::BoxBody;
use http_body_util::{BodyExt, Empty, Full, StreamBody};
use hyper::body::{Body, Bytes, Frame};
use hyper::header::{HeaderName, HeaderValue};
use hyper::{header, HeaderMap, Method, Request, Response, StatusCode};
#[cfg(feature = "runtime-tokio")]
use tokio::io::BufReader;
#[cfg(feature = "runtime-tokio")]
use tokio_util::io::ReaderStream;

use crate::config::{ServerConfiguration, ServerConfigurations};
use crate::get_value;
use crate::runtime::timeout;
#[cfg(feature = "runtime-monoio")]
use crate::util::MonoioFileStreamNoSpawn;
use crate::util::{
  generate_default_error_page, replace_header_placeholders, replace_log_placeholders, sanitize_url, SERVER_SOFTWARE,
};

use ferron_common::modules::{ModuleHandlers, RequestData, SocketData};
use ferron_common::{get_entries, get_entry};

/// Generates an error response
async fn generate_error_response(
  status_code: StatusCode,
  config: &ServerConfiguration,
  headers: &Option<HeaderMap>,
) -> Response<BoxBody<Bytes, std::io::Error>> {
  let bare_body = generate_default_error_page(
    status_code,
    get_value!("server_administrator_email", config).and_then(|v| v.as_str()),
  );
  let mut content_length: Option<u64> = bare_body.len().try_into().ok();
  let mut response_body = Full::new(Bytes::from(bare_body)).map_err(|e| match e {}).boxed();

  if let Some(error_pages) = get_entries!("error_page", config) {
    for error_page in &error_pages.inner {
      if let Some(page_status_code) = error_page.values.first().and_then(|v| v.as_i128()) {
        if page_status_code
          .try_into()
          .ok()
          .and_then(|s| StatusCode::from_u16(s).ok())
          .is_none_or(|s| s != status_code)
        {
          continue;
        }
        if let Some(page_path) = error_page.values.get(1).and_then(|v| v.as_str()) {
          #[cfg(feature = "runtime-monoio")]
          let file = monoio::fs::File::open(page_path).await;
          #[cfg(feature = "runtime-tokio")]
          let file = tokio::fs::File::open(page_path).await;

          let Ok(file) = file else {
            continue;
          };

          // Monoio's `File` doesn't expose `metadata()` on Windows, so we have to spawn a blocking task to obtain the metadata on this platform
          #[cfg(any(feature = "runtime-tokio", all(feature = "runtime-monoio", unix)))]
          let metadata = file.metadata().await;
          #[cfg(all(feature = "runtime-monoio", windows))]
          let metadata = {
            let page_path = page_path.to_owned();
            monoio::spawn_blocking(move || std::fs::metadata(page_path))
              .await
              .unwrap_or(Err(std::io::Error::other(
                "Can't spawn a blocking task to obtain the file metadata",
              )))
          };

          content_length = metadata.ok().map(|m| m.len());

          #[cfg(feature = "runtime-monoio")]
          let file_stream = MonoioFileStreamNoSpawn::new(file, None, content_length);
          #[cfg(feature = "runtime-tokio")]
          let file_stream = ReaderStream::new(BufReader::with_capacity(12800, file));

          let stream_body = StreamBody::new(file_stream.map_ok(Frame::data));
          let boxed_body = stream_body.boxed();

          response_body = boxed_body;

          break;
        }
      }
    }
  }

  let mut response_builder = Response::builder().status(status_code);

  if let Some(headers) = headers {
    let headers_iter = headers.iter();
    for (name, value) in headers_iter {
      if name != header::CONTENT_TYPE && name != header::CONTENT_LENGTH {
        response_builder = response_builder.header(name, value);
      }
    }
  }

  if let Some(content_length) = content_length {
    response_builder = response_builder.header(header::CONTENT_LENGTH, content_length);
  }
  response_builder = response_builder.header(header::CONTENT_TYPE, HeaderValue::from_static("text/html"));

  response_builder.body(response_body).unwrap_or_default()
}

/// Sends a log message formatted according to the Combined Log Format
#[allow(clippy::too_many_arguments)]
async fn log_access(
  loggers: &[Sender<LogMessage>],
  request_parts: &hyper::http::request::Parts,
  socket_data: &SocketData,
  auth_user: Option<&str>,
  status_code: u16,
  content_length: Option<u64>,
  date_format: Option<&str>,
  log_format: Option<&str>,
) {
  let now: DateTime<Local> = Local::now();
  let formatted_time = now.format(date_format.unwrap_or("%d/%b/%Y:%H:%M:%S %z")).to_string();
  let log_message_string = replace_log_placeholders(
    log_format.unwrap_or(
      "{client_ip} - {auth_user} [{timestamp}] \"{method} {path_and_query} {version}\" \
       {status_code} {content_length} \"{header:Referer}\" \"{header:User-Agent}\"",
    ),
    request_parts,
    socket_data,
    auth_user,
    &formatted_time,
    status_code,
    content_length,
  );
  for logger in loggers {
    logger
      .send(LogMessage::new(log_message_string.clone(), false))
      .await
      .unwrap_or_default();
  }
}

/// Helper function to add custom headers to response
#[inline]
fn add_custom_headers(
  response_parts: &mut hyper::http::response::Parts,
  headers_to_add: &HeaderMap,
  headers_to_replace: &HeaderMap,
  headers_to_remove: &[HeaderName],
) {
  for (header_name, header_value) in headers_to_add {
    if !response_parts.headers.contains_key(header_name) {
      response_parts.headers.insert(header_name, header_value.to_owned());
    }
  }

  for (header_name, header_value) in headers_to_replace {
    response_parts.headers.insert(header_name, header_value.to_owned());
  }

  for header_to_remove in headers_to_remove.iter().rev() {
    if response_parts.headers.contains_key(header_to_remove) {
      while response_parts.headers.remove(header_to_remove).is_some() {}
    }
  }
}

/// Helper function to add HTTP/3 Alt-Svc header
#[inline]
fn add_http3_alt_svc_header(response_parts: &mut hyper::http::response::Parts, http3_alt_port: Option<u16>) {
  if let Some(http3_alt_port) = http3_alt_port {
    if let Ok(header_value) = match response_parts.headers.get(header::ALT_SVC) {
      Some(value) => {
        let header_value_old = String::from_utf8_lossy(value.as_bytes());
        let header_value_new = format!("h3=\":{http3_alt_port}\", h3-29=\":{http3_alt_port}\"");

        if header_value_old != header_value_new {
          HeaderValue::from_bytes(format!("{header_value_old}, {header_value_new}").as_bytes())
        } else {
          HeaderValue::from_bytes(header_value_old.as_bytes())
        }
      }
      None => HeaderValue::from_bytes(format!("h3=\":{http3_alt_port}\", h3-29=\":{http3_alt_port}\"").as_bytes()),
    } {
      response_parts.headers.insert(header::ALT_SVC, header_value);
    }
  }
}

/// Helper function to add server header
#[inline]
fn add_server_header(response_parts: &mut hyper::http::response::Parts) {
  response_parts
    .headers
    .insert(header::SERVER, HeaderValue::from_static(SERVER_SOFTWARE));
}

/// Helper function to extract content length for logging
fn extract_content_length(response: &Response<BoxBody<Bytes, std::io::Error>>) -> Option<u64> {
  response
    .headers()
    .get(header::CONTENT_LENGTH)
    .and_then(|header_value| {
      header_value.to_str().ok().and_then(|header_value| {
        header_value
          .parse::<u64>()
          .ok()
          .or_else(|| response.body().size_hint().exact())
      })
    })
    .or_else(|| response.body().size_hint().exact())
}

/// Helper function to build a basic HTML error response without configuration context
fn basic_error_response(status_code: StatusCode) -> Response<BoxBody<Bytes, std::io::Error>> {
  Response::builder()
    .status(status_code)
    .header(header::CONTENT_TYPE, HeaderValue::from_static("text/html"))
    .body(
      Full::new(Bytes::from(generate_default_error_page(status_code, None)))
        .map_err(|e| match e {})
        .boxed(),
    )
    .unwrap_or_default()
}

/// Helper function to build custom header sets for responses
fn build_custom_headers(
  configuration: &ServerConfiguration,
  request_parts: &hyper::http::request::Parts,
) -> (HeaderMap, HeaderMap, Vec<HeaderName>) {
  let mut headers_to_add = HeaderMap::new();
  let mut headers_to_replace = HeaderMap::new();
  let mut headers_to_remove = Vec::new();

  if let Some(custom_headers) = get_entries!("header", configuration) {
    for custom_header in custom_headers.inner.iter().rev() {
      if let Some(header_name) = custom_header.values.first().and_then(|v| v.as_str()) {
        if let Some(header_value) = custom_header.values.get(1).and_then(|v| v.as_str()) {
          if !headers_to_add.contains_key(header_name) {
            if let Ok(header_name) = HeaderName::from_str(header_name) {
              if let Ok(header_value) =
                HeaderValue::from_str(&replace_header_placeholders(header_value, request_parts, None))
              {
                headers_to_add.insert(header_name, header_value);
              }
            }
          }
        }
      }
    }
  }

  if let Some(custom_headers) = get_entries!("header_replace", configuration) {
    for custom_header in custom_headers.inner.iter().rev() {
      if let Some(header_name) = custom_header.values.first().and_then(|v| v.as_str()) {
        if let Some(header_value) = custom_header.values.get(1).and_then(|v| v.as_str()) {
          if let Ok(header_name) = HeaderName::from_str(header_name) {
            if let Ok(header_value) =
              HeaderValue::from_str(&replace_header_placeholders(header_value, request_parts, None))
            {
              headers_to_replace.insert(header_name, header_value);
            }
          }
        }
      }
    }
  }

  if let Some(custom_headers_to_remove) = get_entries!("header_remove", configuration) {
    for custom_header in custom_headers_to_remove.inner.iter().rev() {
      if let Some(header_name) = custom_header.values.first().and_then(|v| v.as_str()) {
        if let Ok(header_name) = HeaderName::from_str(header_name) {
          headers_to_remove.push(header_name);
        }
      }
    }
  }

  (headers_to_add, headers_to_replace, headers_to_remove)
}

/// Helper function to apply all response headers and log if needed
#[allow(clippy::too_many_arguments)]
async fn finalize_response_and_log(
  response: Response<BoxBody<Bytes, std::io::Error>>,
  http3_alt_port: Option<u16>,
  headers_to_add: HeaderMap,
  headers_to_replace: HeaderMap,
  headers_to_remove: Vec<HeaderName>,
  loggers: &[Sender<LogMessage>],
  request_parts: &Option<hyper::http::request::Parts>,
  socket_data: &SocketData,
  latest_auth_data: Option<&str>,
  date_format: Option<&str>,
  log_format: Option<&str>,
) -> Response<BoxBody<Bytes, std::io::Error>> {
  let (mut response_parts, response_body) = response.into_parts();

  add_custom_headers(
    &mut response_parts,
    &headers_to_add,
    &headers_to_replace,
    &headers_to_remove,
  );
  add_http3_alt_svc_header(&mut response_parts, http3_alt_port);
  add_server_header(&mut response_parts);

  let response = Response::from_parts(response_parts, response_body);

  if let Some(request_parts) = request_parts {
    if !loggers.is_empty() {
      log_access(
        loggers,
        request_parts,
        socket_data,
        latest_auth_data,
        response.status().as_u16(),
        extract_content_length(&response),
        date_format,
        log_format,
      )
      .await;
    }
  }

  response
}

/// Helper function to finalize a basic error response with logging and default headers
#[allow(clippy::too_many_arguments)]
async fn finalize_basic_error_response(
  response: Response<BoxBody<Bytes, std::io::Error>>,
  request_parts: hyper::http::request::Parts,
  http3_alt_port: Option<u16>,
  loggers: &[Sender<LogMessage>],
  socket_data: &SocketData,
  date_format: Option<&str>,
  log_format: Option<&str>,
) -> Response<BoxBody<Bytes, std::io::Error>> {
  let request_parts = Some(request_parts);
  finalize_response_and_log(
    response,
    http3_alt_port,
    HeaderMap::new(),
    HeaderMap::new(),
    Vec::new(),
    loggers,
    &request_parts,
    socket_data,
    None,
    date_format,
    log_format,
  )
  .await
}

/// Helper function to execute response modifying handlers
#[allow(clippy::too_many_arguments)]
#[inline]
async fn execute_response_modifying_handlers(
  mut response: Response<BoxBody<Bytes, std::io::Error>>,
  mut executed_handlers: Vec<Box<dyn ModuleHandlers>>,
  configuration: &ServerConfiguration,
  http3_alt_port: Option<u16>,
  headers_to_add: HeaderMap,
  headers_to_replace: HeaderMap,
  headers_to_remove: Vec<HeaderName>,
  loggers: &[Sender<LogMessage>],
  request_parts: &Option<hyper::http::request::Parts>,
  socket_data: &SocketData,
  latest_auth_data: Option<&str>,
  date_format: Option<&str>,
  log_format: Option<&str>,
  metrics_sender: MetricsMultiSender,
  metrics_enabled: bool,
  traces_senders: Vec<Sender<TraceSignal>>,
  traces_enabled: bool,
  timeout_instant: std::time::Instant,
  timeout_duration: Option<std::time::Duration>,
) -> Result<Result<Response<BoxBody<Bytes, std::io::Error>>, Response<BoxBody<Bytes, std::io::Error>>>, anyhow::Error> {
  while let Some(mut executed_handler) = executed_handlers.pop() {
    if traces_enabled {
      for trace_sender in &traces_senders {
        trace_sender
          .send(TraceSignal::StartSpan(format!(
            "{}::response_modifying_handler",
            executed_handler.get_name()
          )))
          .await
          .unwrap_or_default();
      }
    }
    let (response_status, is_timeout) = if let Some(timeout_duration) = &timeout_duration {
      let elapsed = timeout_instant.elapsed();
      if let Some(timeout_cur_duration) = timeout_duration.checked_sub(elapsed) {
        match timeout(
          timeout_cur_duration,
          executed_handler.response_modifying_handler(response),
        )
        .await
        {
          Ok(result) => (result, false),
          Err(_) => (Err(anyhow::anyhow!("The client or server has timed out").into()), true),
        }
      } else {
        (Err(anyhow::anyhow!("The client or server has timed out").into()), true)
      }
    } else {
      (executed_handler.response_modifying_handler(response).await, false)
    };
    if traces_enabled {
      for trace_sender in &traces_senders {
        trace_sender
          .send(TraceSignal::EndSpan(
            format!("{}::response_modifying_handler", executed_handler.get_name()),
            response_status.as_ref().err().map(|e| e.to_string()),
          ))
          .await
          .unwrap_or_default();
      }
    }
    if is_timeout {
      if metrics_enabled {
        while let Some(mut executed_handler) = executed_handlers.pop() {
          executed_handler.metric_data_after_handler(&metrics_sender).await;
        }
      }
      Err(anyhow::anyhow!("The client or server has timed out"))?;
    }
    response = match response_status {
      Ok(response) => response,
      Err(err) => {
        for logger in loggers {
          logger
            .send(LogMessage::new(
              format!("Unexpected error while serving a request: {err}"),
              true,
            ))
            .await
            .unwrap_or_default();
        }

        let error_response = generate_error_response(StatusCode::INTERNAL_SERVER_ERROR, configuration, &None).await;

        let final_response = finalize_response_and_log(
          error_response,
          http3_alt_port,
          headers_to_add,
          headers_to_replace,
          headers_to_remove,
          loggers,
          request_parts,
          socket_data,
          latest_auth_data,
          date_format,
          log_format,
        )
        .await;

        if metrics_enabled {
          while let Some(mut executed_handler) = executed_handlers.pop() {
            executed_handler.metric_data_after_handler(&metrics_sender).await;
          }
        }

        return Ok(Err(final_response));
      }
    };
    if metrics_enabled {
      executed_handler.metric_data_after_handler(&metrics_sender).await;
    }
  }
  Ok(Ok(response))
}

/// Helper function to apply response headers, execute response modifiers, and log access if needed
#[allow(clippy::too_many_arguments)]
async fn finalize_with_modifying_handlers(
  response: Response<BoxBody<Bytes, std::io::Error>>,
  executed_handlers: Vec<Box<dyn ModuleHandlers>>,
  configuration: &ServerConfiguration,
  http3_alt_port: Option<u16>,
  headers_to_add: HeaderMap,
  headers_to_replace: HeaderMap,
  headers_to_remove: Vec<HeaderName>,
  log_request_parts: &Option<hyper::http::request::Parts>,
  socket_data: &SocketData,
  latest_auth_data: Option<&str>,
  log_date_format: Option<&str>,
  log_format: Option<&str>,
  metrics_sender: MetricsMultiSender,
  metrics_enabled: bool,
  traces_senders: Vec<Sender<TraceSignal>>,
  traces_enabled: bool,
  timeout_instant: std::time::Instant,
  timeout_duration: Option<std::time::Duration>,
) -> Result<Response<BoxBody<Bytes, std::io::Error>>, anyhow::Error> {
  let (mut response_parts, response_body) = response.into_parts();
  add_custom_headers(
    &mut response_parts,
    &headers_to_add,
    &headers_to_replace,
    &headers_to_remove,
  );
  add_http3_alt_svc_header(&mut response_parts, http3_alt_port);
  add_server_header(&mut response_parts);

  let response = Response::from_parts(response_parts, response_body);

  match execute_response_modifying_handlers(
    response,
    executed_handlers,
    configuration,
    http3_alt_port,
    headers_to_add,
    headers_to_replace,
    headers_to_remove,
    &configuration.observability.log_channels,
    log_request_parts,
    socket_data,
    latest_auth_data,
    log_date_format,
    log_format,
    metrics_sender,
    metrics_enabled,
    traces_senders,
    traces_enabled,
    timeout_instant,
    timeout_duration,
  )
  .await?
  {
    Ok(response) => {
      if let Some(request_parts) = log_request_parts.as_ref() {
        log_access(
          &configuration.observability.log_channels,
          request_parts,
          socket_data,
          latest_auth_data,
          response.status().as_u16(),
          extract_content_length(&response),
          log_date_format,
          log_format,
        )
        .await;
      }
      Ok(response)
    }
    Err(error_response) => Ok(error_response),
  }
}

#[allow(clippy::type_complexity)]
#[allow(clippy::result_large_err)]
fn sanitize_url_request(
  configuration: &ServerConfiguration,
  mut request: Request<BoxBody<Bytes, std::io::Error>>,
) -> Result<
  (Request<BoxBody<Bytes, std::io::Error>>, bool),
  (
    Request<BoxBody<Bytes, std::io::Error>>,
    Box<dyn std::error::Error + Send + Sync>,
  ),
> {
  let url_pathname = request.uri().path();
  let sanitized_url_pathname = match sanitize_url(
    url_pathname,
    get_value!("allow_double_slashes", configuration)
      .and_then(|v| v.as_bool())
      .unwrap_or(false),
  ) {
    Ok(sanitized_url_pathname) => sanitized_url_pathname,
    Err(err) => {
      return Err((request, err.into()));
    }
  };

  if sanitized_url_pathname != url_pathname {
    let (mut parts, body) = request.into_parts();
    let orig_uri = parts.uri.clone();
    let mut url_parts = parts.uri.into_parts();
    url_parts.path_and_query = Some(
      match format!(
        "{}{}",
        sanitized_url_pathname,
        url_parts
          .path_and_query
          .as_ref()
          .and_then(|pq| pq.query())
          .map_or("".to_string(), |q| format!("?{q}"))
      )
      .parse()
      {
        Ok(path_and_query) => path_and_query,
        Err(e) => {
          parts.uri = orig_uri;
          request = Request::from_parts(parts, body);
          return Err((request, e.into()));
        }
      },
    );
    parts.uri = match hyper::Uri::from_parts(url_parts) {
      Ok(uri) => uri,
      Err(e) => {
        parts.uri = orig_uri;
        request = Request::from_parts(parts, body);
        return Err((request, e.into()));
      }
    };
    request = Request::from_parts(parts, body);
    return Ok((request, true));
  }
  Ok((request, false))
}

/// The HTTP request handler, with timeout
#[allow(clippy::too_many_arguments)]
pub async fn request_handler(
  mut request: Request<BoxBody<Bytes, std::io::Error>>,
  client_address: SocketAddr,
  server_address: SocketAddr,
  encrypted: bool,
  configurations: Arc<ServerConfigurations>,
  http3_alt_port: Option<u16>,
  acme_http_01_resolvers: Arc<tokio::sync::RwLock<Vec<crate::acme::Http01DataLock>>>,
  proxy_protocol_client_address: Option<SocketAddr>,
  proxy_protocol_server_address: Option<SocketAddr>,
) -> Result<Response<BoxBody<Bytes, std::io::Error>>, anyhow::Error> {
  // Global configuration
  let global_configuration = configurations.find_global_configuration();
  let global_loggers: &[Sender<LogMessage>] = global_configuration
    .as_ref()
    .map_or(&[], |c| &*c.observability.log_channels);
  let global_log_date_format = global_configuration
    .as_deref()
    .and_then(|c| get_value!("log_date_format", c))
    .and_then(|v| v.as_str());
  let global_log_format = global_configuration
    .as_deref()
    .and_then(|c| get_value!("log_format", c))
    .and_then(|v| v.as_str());

  // Request timeout
  let timeout_from_config = global_configuration
    .as_deref()
    .and_then(|c| get_entry!("timeout", c))
    .and_then(|e| e.values.last());
  let timeout_duration = if timeout_from_config.is_some_and(|v| v.is_null()) {
    None
  } else {
    let timeout_millis = timeout_from_config.and_then(|v| v.as_i128()).unwrap_or(300000) as u64;
    Some(Duration::from_millis(timeout_millis))
  };
  let timeout_instant = std::time::Instant::now();

  // Normalize HTTP/2 and HTTP/3 request objects
  if matches!(request.version(), hyper::Version::HTTP_2 | hyper::Version::HTTP_3) {
    // Set "Host" request header for HTTP/2 and HTTP/3 connections
    if let Some(authority) = request.uri().authority() {
      let authority = authority.to_owned();
      let headers = request.headers_mut();
      if !headers.contains_key(header::HOST) {
        if let Ok(authority_value) = HeaderValue::from_bytes(authority.as_str().as_bytes()) {
          headers.append(header::HOST, authority_value);
        }
      }
    }

    // Normalize the Cookie header for HTTP/2 and HTTP/3
    let mut cookie_normalized = String::new();
    let mut cookie_set = false;
    let headers = request.headers_mut();
    for cookie in headers.get_all(header::COOKIE) {
      if let Ok(cookie) = cookie.to_str() {
        if cookie_set {
          cookie_normalized.push_str("; ");
        }
        cookie_set = true;
        cookie_normalized.push_str(cookie);
      }
    }
    if cookie_set {
      if let Ok(cookie_value) = HeaderValue::from_bytes(cookie_normalized.as_bytes()) {
        headers.insert(header::COOKIE, cookie_value);
      }
    }
  }

  // Construct socket data
  let mut socket_data = SocketData {
    remote_addr: proxy_protocol_client_address.unwrap_or(client_address),
    local_addr: proxy_protocol_server_address.unwrap_or(server_address),
    encrypted,
  };

  // Sanitize "Host" header
  let host_header_option = request.headers().get(header::HOST);
  if let Some(header_data) = host_header_option {
    match header_data.to_str() {
      Ok(host_header) => {
        let host_header_lower_case = host_header.to_lowercase();
        let host_header_without_dot = host_header_lower_case
          .strip_suffix('.')
          .unwrap_or(host_header_lower_case.as_str());
        if host_header_without_dot != host_header {
          let host_header_value = match HeaderValue::from_str(host_header_without_dot) {
            Ok(host_header_value) => host_header_value,
            Err(err) => {
              for logger in global_loggers {
                logger
                  .send(LogMessage::new(format!("Host header sanitation error: {err}"), true))
                  .await
                  .unwrap_or_default();
              }
              let response = basic_error_response(StatusCode::BAD_REQUEST);
              let (request_parts, _) = request.into_parts();
              return Ok(
                finalize_basic_error_response(
                  response,
                  request_parts,
                  http3_alt_port,
                  global_loggers,
                  &socket_data,
                  global_log_date_format,
                  global_log_format,
                )
                .await,
              );
            }
          };

          request.headers_mut().insert(header::HOST, host_header_value);
        }
      }
      Err(err) => {
        for logger in global_loggers {
          logger
            .send(LogMessage::new(format!("Host header sanitation error: {err}"), true))
            .await
            .unwrap_or_default();
        }
        let response = basic_error_response(StatusCode::BAD_REQUEST);
        let (request_parts, _) = request.into_parts();
        return Ok(
          finalize_basic_error_response(
            response,
            request_parts,
            http3_alt_port,
            global_loggers,
            &socket_data,
            global_log_date_format,
            global_log_format,
          )
          .await,
        );
      }
    }
  };

  let hostname_determinant = request.headers().get(header::HOST).and_then(|value| {
    value.to_str().ok().map(|h| {
      h.rsplit_once(':')
        .and_then(|(left, right)| {
          if right.parse::<u16>().is_ok() {
            Some(left.to_string())
          } else {
            None
          }
        })
        .unwrap_or_else(|| h.to_string())
    })
  });

  let (request_parts, request_body) = request.into_parts();
  let mut log_request_parts = if global_configuration
    .as_ref()
    .is_some_and(|c| !c.observability.log_channels.is_empty())
  {
    Some(request_parts.clone())
  } else {
    None
  };
  let request = Request::from_parts(request_parts, request_body);

  // Find the server configuration
  let (request_parts, request_body) = request.into_parts();
  let configuration_option =
    configurations.find_configuration(&request_parts, hostname_determinant.as_deref(), &socket_data);
  let mut request = Request::from_parts(request_parts, request_body);
  let mut configuration_error_handler_lookup = match configuration_option
    .and_then(|c| c.ok_or_else(|| anyhow::anyhow!("No matching configuration found").into_boxed_dyn_error()))
  {
    Ok(configuration) => configuration,
    Err(err) => {
      for logger in global_loggers {
        logger
          .send(LogMessage::new(
            format!("Cannot determine server configuration: {err}"),
            true,
          ))
          .await
          .unwrap_or_default()
      }
      let response = basic_error_response(StatusCode::INTERNAL_SERVER_ERROR);
      let (request_parts, _) = request.into_parts();
      return Ok(
        finalize_basic_error_response(
          response,
          request_parts,
          http3_alt_port,
          global_loggers,
          &socket_data,
          global_log_date_format,
          global_log_format,
        )
        .await,
      );
    }
  };
  let mut configuration = match configuration_error_handler_lookup.get_default().cloned() {
    Some(configuration) => configuration,
    None => {
      for logger in global_loggers {
        logger
          .send(LogMessage::new(
            "Cannot determine server configuration: No matching configuration found".to_string(),
            true,
          ))
          .await
          .unwrap_or_default()
      }
      let response = basic_error_response(StatusCode::INTERNAL_SERVER_ERROR);
      let (request_parts, _) = request.into_parts();
      return Ok(
        finalize_basic_error_response(
          response,
          request_parts,
          http3_alt_port,
          global_loggers,
          &socket_data,
          global_log_date_format,
          global_log_format,
        )
        .await,
      );
    }
  };

  // Determine the log formats
  let mut log_date_format = get_value!("log_date_format", configuration).and_then(|v| v.as_str());
  let mut log_format = get_value!("log_format", configuration).and_then(|v| v.as_str());

  // Clone the log request parts if logging is enabled and request parts are not already cloned
  if !configuration.observability.log_channels.is_empty() && log_request_parts.is_none() {
    let (request_parts, request_body) = request.into_parts();
    log_request_parts = Some(request_parts.clone());
    request = Request::from_parts(request_parts, request_body);
  }

  // Sanitize the URL, if the URL sanitizer is enabled
  if !get_value!("disable_url_sanitizer", configuration)
    .and_then(|v| v.as_bool())
    .unwrap_or(false)
  {
    request = match sanitize_url_request(&configuration, request) {
      Ok((mut request, was_dirty)) => {
        if was_dirty {
          let (parts, body) = request.into_parts();
          let configuration_option =
            configurations.find_configuration(&parts, hostname_determinant.as_deref(), &socket_data);
          request = Request::from_parts(parts, body);
          match configuration_option {
            Ok(Some(new_configuration)) => {
              if let Some(new_config2) = new_configuration.get_default().cloned() {
                configuration_error_handler_lookup = new_configuration;
                configuration = new_config2;
                log_date_format = get_value!("log_date_format", configuration).and_then(|v| v.as_str());
                log_format = get_value!("log_format", configuration).and_then(|v| v.as_str());
              }
            }
            Ok(None) => {}
            Err(err) => {
              for logger in &configuration.observability.log_channels {
                logger
                  .send(LogMessage::new(
                    format!("Cannot determine server configuration: {err}"),
                    true,
                  ))
                  .await
                  .unwrap_or_default();
              }
              let response = generate_error_response(StatusCode::BAD_REQUEST, &configuration, &None).await;

              let (request_parts, _) = request.into_parts();
              let (headers_to_add, headers_to_replace, headers_to_remove) =
                build_custom_headers(&configuration, &request_parts);

              return Ok(
                finalize_response_and_log(
                  response,
                  http3_alt_port,
                  headers_to_add,
                  headers_to_replace,
                  headers_to_remove,
                  &configuration.observability.log_channels,
                  &log_request_parts,
                  &socket_data,
                  None,
                  log_date_format,
                  log_format,
                )
                .await,
              );
            }
          }
        }
        request
      }
      Err((request, error)) => {
        for logger in &configuration.observability.log_channels {
          logger
            .send(LogMessage::new(format!("URL sanitation error: {error}"), true))
            .await
            .unwrap_or_default();
        }

        let response = generate_error_response(StatusCode::BAD_REQUEST, &configuration, &None).await;

        let (parts, _) = request.into_parts();
        let (headers_to_add, headers_to_replace, headers_to_remove) = build_custom_headers(&configuration, &parts);

        return Ok(
          finalize_response_and_log(
            response,
            http3_alt_port,
            headers_to_add,
            headers_to_replace,
            headers_to_remove,
            &configuration.observability.log_channels,
            &log_request_parts,
            &socket_data,
            None,
            log_date_format,
            log_format,
          )
          .await,
        );
      }
    };
  }

  let (request_parts, request_body) = request.into_parts();
  let (headers_to_add, headers_to_replace, headers_to_remove) = build_custom_headers(&configuration, &request_parts);
  let mut request = Request::from_parts(request_parts, request_body);

  if request.uri().path() == "*" {
    let response = match request.method() {
      &Method::OPTIONS => Response::builder()
        .status(StatusCode::NO_CONTENT)
        .header(header::ALLOW, HeaderValue::from_static("GET, POST, HEAD, OPTIONS"))
        .body(Empty::new().map_err(|e| match e {}).boxed())
        .unwrap_or_default(),
      _ => {
        let mut header_map = HeaderMap::new();
        header_map.insert(header::ALLOW, HeaderValue::from_static("GET, POST, HEAD, OPTIONS"));
        generate_error_response(StatusCode::BAD_REQUEST, &configuration, &Some(header_map)).await
      }
    };
    return Ok(
      finalize_response_and_log(
        response,
        http3_alt_port,
        headers_to_add,
        headers_to_replace,
        headers_to_remove,
        &configuration.observability.log_channels,
        &log_request_parts,
        &socket_data,
        None,
        log_date_format,
        log_format,
      )
      .await,
    );
  }

  // HTTP-01 ACME challenge for automatic TLS
  let acme_http_01_resolvers_inner = acme_http_01_resolvers.read().await;
  if !acme_http_01_resolvers_inner.is_empty() {
    if let Some(challenge_token) = request.uri().path().strip_prefix("/.well-known/acme-challenge/") {
      for acme_http01_resolver in &*acme_http_01_resolvers_inner {
        if let Some(http01_acme_data) = &*acme_http01_resolver.read().await {
          let acme_response = http01_acme_data.1.clone();
          if challenge_token == http01_acme_data.0 {
            let response = Response::builder()
              .status(StatusCode::OK)
              .header(
                header::CONTENT_TYPE,
                HeaderValue::from_static("application/octet-stream"),
              )
              .body(Full::new(Bytes::from(acme_response)).map_err(|e| match e {}).boxed())
              .unwrap_or_default();

            return Ok(
              finalize_response_and_log(
                response,
                http3_alt_port,
                headers_to_add,
                headers_to_replace,
                headers_to_remove,
                &configuration.observability.log_channels,
                &log_request_parts,
                &socket_data,
                None,
                log_date_format,
                log_format,
              )
              .await,
            );
          }
        }
      }
    }
  };
  drop(acme_http_01_resolvers_inner);

  let mut error_logger = if !configuration.observability.log_channels.is_empty() {
    ErrorLogger::new_multiple(configuration.observability.log_channels.clone())
  } else {
    ErrorLogger::without_logger()
  };
  let mut metrics_enabled = !configuration.observability.metric_channels.is_empty();
  let mut metrics_sender = if metrics_enabled {
    MetricsMultiSender::new_multiple(configuration.observability.metric_channels.clone())
  } else {
    MetricsMultiSender::without_sender()
  };
  let mut traces_enabled = !configuration.observability.trace_channels.is_empty();
  let mut traces_senders = if traces_enabled {
    let mut traces_senders = Vec::with_capacity(configuration.observability.trace_channels.len());
    for channel in &configuration.observability.trace_channels {
      channel.0.send(()).await.unwrap_or_default();
      if let Ok(channel2) = channel.1.recv().await {
        traces_senders.push(channel2);
      }
    }
    traces_senders
  } else {
    vec![]
  };

  // Obtain module handlers
  let mut module_handlers = Vec::with_capacity(configuration.modules.len());
  for module in &configuration.modules {
    module_handlers.push(module.get_module_handlers());
  }

  // Execute modules!
  request.extensions_mut().insert(RequestData {
    auth_user: None,
    original_url: None,
    error_status_code: None,
  });
  let mut executed_handlers = Vec::new();
  let (request_parts, request_body) = request.into_parts();
  let request_parts_cloned = if configuration_error_handler_lookup.has_status_codes() {
    let mut request_parts_cloned = request_parts.clone();
    request_parts_cloned
      .headers
      .insert(header::CONTENT_LENGTH, HeaderValue::from_static("0"));
    Some(request_parts_cloned)
  } else {
    // If the error configuration is not specified, don't clone the request parts to improve performance
    None
  };
  let mut request = Request::from_parts(request_parts, request_body);
  let mut latest_auth_data = None;
  let mut is_error_handler = false;
  let mut handlers_iter: Box<dyn Iterator<Item = Box<dyn ModuleHandlers>>> = Box::new(module_handlers.into_iter());
  while let Some(mut handlers) = handlers_iter.next() {
    if metrics_enabled {
      handlers
        .metric_data_before_handler(&request, &socket_data, &metrics_sender)
        .await;
    }

    if traces_enabled {
      for trace_sender in &traces_senders {
        trace_sender
          .send(TraceSignal::StartSpan(format!(
            "{}::request_handler",
            handlers.get_name()
          )))
          .await
          .unwrap_or_default();
      }
    }

    let (response_result, is_timeout) = if let Some(timeout_duration) = &timeout_duration {
      let elapsed = timeout_instant.elapsed();
      if let Some(timeout_cur_duration) = timeout_duration.checked_sub(elapsed) {
        match timeout(
          timeout_cur_duration,
          handlers.request_handler(request, &configuration, &socket_data, &error_logger),
        )
        .await
        {
          Ok(result) => (result, false),
          Err(_) => (Err(anyhow::anyhow!("The client or server has timed out").into()), true),
        }
      } else {
        (Err(anyhow::anyhow!("The client or server has timed out").into()), true)
      }
    } else {
      (
        handlers
          .request_handler(request, &configuration, &socket_data, &error_logger)
          .await,
        false,
      )
    };

    if traces_enabled {
      for trace_sender in &traces_senders {
        trace_sender
          .send(TraceSignal::EndSpan(
            format!("{}::request_handler", handlers.get_name()),
            response_result.as_ref().err().map(|e| e.to_string()),
          ))
          .await
          .unwrap_or_default();
      }
    }

    executed_handlers.push(handlers);

    if is_timeout {
      if metrics_enabled {
        while let Some(mut executed_handler) = executed_handlers.pop() {
          executed_handler.metric_data_after_handler(&metrics_sender).await;
        }
      }
      Err(anyhow::anyhow!("The client or server has timed out"))?;
    }

    match response_result {
      Ok(response) => {
        let status = response.response_status;
        let headers = response.response_headers;
        let new_remote_address = response.new_remote_address;
        let request_option = response.request;
        let response = response.response;
        let request_extensions = request_option
          .as_ref()
          .and_then(|r| r.extensions().get::<RequestData>());
        if let Some(request_extensions) = request_extensions {
          latest_auth_data = request_extensions.auth_user.clone();
        }
        if let Some(new_remote_address) = new_remote_address {
          socket_data.remote_addr = new_remote_address;
        };

        match response {
          Some(response) => {
            return finalize_with_modifying_handlers(
              response,
              executed_handlers,
              &configuration,
              http3_alt_port,
              headers_to_add,
              headers_to_replace,
              headers_to_remove,
              &log_request_parts,
              &socket_data,
              latest_auth_data.as_deref(),
              log_date_format,
              log_format,
              metrics_sender,
              metrics_enabled,
              traces_senders,
              traces_enabled,
              timeout_instant,
              timeout_duration,
            )
            .await;
          }
          None => match status {
            Some(status) => {
              if !is_error_handler {
                if let Some(error_configuration) = configuration_error_handler_lookup.get(status.as_u16()).cloned() {
                  let request_option = if let Some(request) = request_option {
                    Some(request)
                  } else {
                    request_parts_cloned.clone().map(|request_parts_cloned| {
                      Request::from_parts(request_parts_cloned, Empty::new().map_err(|e| match e {}).boxed())
                    })
                  };
                  if let Some(request_cloned) = request_option {
                    configuration = error_configuration;
                    let mut module_handlers = Vec::with_capacity(configuration.modules.len());
                    for module in &configuration.modules {
                      module_handlers.push(module.get_module_handlers());
                    }
                    handlers_iter = Box::new(module_handlers.into_iter());
                    if metrics_enabled {
                      while let Some(mut executed_handler) = executed_handlers.pop() {
                        executed_handler.metric_data_after_handler(&metrics_sender).await;
                      }
                    }
                    executed_handlers = Vec::new();
                    request = request_cloned;
                    if let Some(request_data) = request.extensions_mut().get_mut::<RequestData>() {
                      request_data.error_status_code = Some(status);
                    }
                    is_error_handler = true;
                    log_date_format = get_value!("log_date_format", configuration).and_then(|v| v.as_str());
                    log_format = get_value!("log_format", configuration).and_then(|v| v.as_str());
                    error_logger = if !configuration.observability.log_channels.is_empty() {
                      ErrorLogger::new_multiple(configuration.observability.log_channels.clone())
                    } else {
                      ErrorLogger::without_logger()
                    };
                    metrics_enabled = !configuration.observability.metric_channels.is_empty();
                    metrics_sender = if metrics_enabled {
                      MetricsMultiSender::new_multiple(configuration.observability.metric_channels.clone())
                    } else {
                      MetricsMultiSender::without_sender()
                    };
                    traces_enabled = !configuration.observability.trace_channels.is_empty();
                    traces_senders = if traces_enabled {
                      let mut traces_senders = Vec::with_capacity(configuration.observability.trace_channels.len());
                      for channel in &configuration.observability.trace_channels {
                        channel.0.send(()).await.unwrap_or_default();
                        if let Ok(channel2) = channel.1.recv().await {
                          traces_senders.push(channel2);
                        }
                      }
                      traces_senders
                    } else {
                      vec![]
                    };
                    continue;
                  }
                }
              }
              let response = generate_error_response(status, &configuration, &headers).await;
              return finalize_with_modifying_handlers(
                response,
                executed_handlers,
                &configuration,
                http3_alt_port,
                headers_to_add,
                headers_to_replace,
                headers_to_remove,
                &log_request_parts,
                &socket_data,
                latest_auth_data.as_deref(),
                log_date_format,
                log_format,
                metrics_sender,
                metrics_enabled,
                traces_senders,
                traces_enabled,
                timeout_instant,
                timeout_duration,
              )
              .await;
            }
            None => match request_option {
              Some(request_obtained) => {
                request = request_obtained;
                continue;
              }
              None => {
                break;
              }
            },
          },
        }
      }
      Err(err) => {
        let response = generate_error_response(StatusCode::INTERNAL_SERVER_ERROR, &configuration, &None).await;

        for logger in &configuration.observability.log_channels {
          logger
            .send(LogMessage::new(
              format!("Unexpected error while serving a request: {err}"),
              true,
            ))
            .await
            .unwrap_or_default();
        }

        return finalize_with_modifying_handlers(
          response,
          executed_handlers,
          &configuration,
          http3_alt_port,
          headers_to_add,
          headers_to_replace,
          headers_to_remove,
          &log_request_parts,
          &socket_data,
          latest_auth_data.as_deref(),
          log_date_format,
          log_format,
          metrics_sender,
          metrics_enabled,
          traces_senders,
          traces_enabled,
          timeout_instant,
          timeout_duration,
        )
        .await;
      }
    }
  }

  let response = generate_error_response(StatusCode::NOT_FOUND, &configuration, &None).await;

  finalize_with_modifying_handlers(
    response,
    executed_handlers,
    &configuration,
    http3_alt_port,
    headers_to_add,
    headers_to_replace,
    headers_to_remove,
    &log_request_parts,
    &socket_data,
    latest_auth_data.as_deref(),
    log_date_format,
    log_format,
    metrics_sender,
    metrics_enabled,
    traces_senders,
    traces_enabled,
    timeout_instant,
    timeout_duration,
  )
  .await
}