chore: change mod name

rpxy-lib/src/message_handler/handler_main.rs

@@ -0,0 +1,253 @@
+use super::{
+  http_log::HttpMessageLog,
+  http_result::{HttpError, HttpResult},
+  synthetic_response::{secure_redirection_response, synthetic_error_response},
+  utils_headers::*,
+  utils_request::InspectParseHost,
+};
+use crate::{
+  backend::{BackendAppManager, LoadBalanceContext},
+  crypto::CryptoSource,
+  error::*,
+  globals::Globals,
+  hyper_ext::body::{BoxBody, IncomingLike, IncomingOr},
+  log::*,
+  name_exp::ServerName,
+};
+use derive_builder::Builder;
+use http::{Request, Response, StatusCode};
+use hyper_util::rt::TokioIo;
+use std::{net::SocketAddr, sync::Arc};
+use tokio::io::copy_bidirectional;
+
+#[allow(dead_code)]
+#[derive(Debug)]
+/// Context object to handle sticky cookies at HTTP message handler
+pub(super) struct HandlerContext {
+  #[cfg(feature = "sticky-cookie")]
+  pub(super) context_lb: Option<LoadBalanceContext>,
+  #[cfg(not(feature = "sticky-cookie"))]
+  pub(super) context_lb: Option<()>,
+}
+
+#[derive(Clone, Builder)]
+/// HTTP message handler for requests from clients and responses from backend applications,
+/// responsible to manipulate and forward messages to upstream backends and downstream clients.
+// pub struct HttpMessageHandler<T, U>
+pub struct HttpMessageHandler<U>
+where
+  // T: Connect + Clone + Sync + Send + 'static,
+  U: CryptoSource + Clone,
+{
+  // forwarder: Arc<Forwarder<T>>,
+  pub(super) globals: Arc<Globals>,
+  app_manager: Arc<BackendAppManager<U>>,
+}
+
+impl<U> HttpMessageHandler<U>
+where
+  // T: Connect + Clone + Sync + Send + 'static,
+  U: CryptoSource + Clone,
+{
+  /// Handle incoming request message from a client.
+  /// Responsible to passthrough responses from backend applications or generate synthetic error responses.
+  pub async fn handle_request(
+    &self,
+    req: Request<IncomingOr<IncomingLike>>,
+    client_addr: SocketAddr, // For access control
+    listen_addr: SocketAddr,
+    tls_enabled: bool,
+    tls_server_name: Option<ServerName>,
+  ) -> RpxyResult<Response<IncomingOr<BoxBody>>> {
+    // preparing log data
+    let mut log_data = HttpMessageLog::from(&req);
+    log_data.client_addr(&client_addr);
+
+    let http_result = self
+      .handle_request_inner(
+        &mut log_data,
+        req,
+        client_addr,
+        listen_addr,
+        tls_enabled,
+        tls_server_name,
+      )
+      .await;
+
+    // passthrough or synthetic response
+    match http_result {
+      Ok(v) => {
+        log_data.status_code(&v.status()).output();
+        Ok(v)
+      }
+      Err(e) => {
+        debug!("{e}");
+        let code = StatusCode::from(e);
+        log_data.status_code(&code).output();
+        synthetic_error_response(code)
+      }
+    }
+  }
+
+  /// Handle inner with no synthetic error response.
+  /// Synthetic response is generated by caller.
+  async fn handle_request_inner(
+    &self,
+    log_data: &mut HttpMessageLog,
+    mut req: Request<IncomingOr<IncomingLike>>,
+    client_addr: SocketAddr, // For access control
+    listen_addr: SocketAddr,
+    tls_enabled: bool,
+    tls_server_name: Option<ServerName>,
+  ) -> HttpResult<Response<IncomingOr<BoxBody>>> {
+    // Here we start to inspect and parse with server_name
+    let server_name = req
+      .inspect_parse_host()
+      .map(|v| ServerName::from(v.as_slice()))
+      .map_err(|_e| HttpError::InvalidHostInRequestHeader)?;
+
+    // check consistency of between TLS SNI and HOST/Request URI Line.
+    #[allow(clippy::collapsible_if)]
+    if tls_enabled && self.globals.proxy_config.sni_consistency {
+      if server_name != tls_server_name.unwrap_or_default() {
+        return Err(HttpError::SniHostInconsistency);
+      }
+    }
+    // Find backend application for given server_name, and drop if incoming request is invalid as request.
+    let backend_app = match self.app_manager.apps.get(&server_name) {
+      Some(backend_app) => backend_app,
+      None => {
+        let Some(default_server_name) = &self.app_manager.default_server_name else {
+          return Err(HttpError::NoMatchingBackendApp);
+        };
+        debug!("Serving by default app");
+        self.app_manager.apps.get(default_server_name).unwrap()
+      }
+    };
+
+    // Redirect to https if !tls_enabled and redirect_to_https is true
+    if !tls_enabled && backend_app.https_redirection.unwrap_or(false) {
+      debug!(
+        "Redirect to secure connection: {}",
+        <&ServerName as TryInto<String>>::try_into(&backend_app.server_name).unwrap_or_default()
+      );
+      return secure_redirection_response(&backend_app.server_name, self.globals.proxy_config.https_port, &req);
+    }
+
+    // Find reverse proxy for given path and choose one of upstream host
+    // Longest prefix match
+    let path = req.uri().path();
+    let Some(upstream_candidates) = backend_app.path_manager.get(path) else {
+      return Err(HttpError::NoUpstreamCandidates);
+    };
+
+    // Upgrade in request header
+    let upgrade_in_request = extract_upgrade(req.headers());
+    let request_upgraded = req.extensions_mut().remove::<hyper::upgrade::OnUpgrade>();
+
+    // Build request from destination information
+    let _context = match self.generate_request_forwarded(
+      &client_addr,
+      &listen_addr,
+      &mut req,
+      &upgrade_in_request,
+      upstream_candidates,
+      tls_enabled,
+    ) {
+      Err(e) => {
+        error!("Failed to generate upstream request for backend application: {}", e);
+        return Err(HttpError::FailedToGenerateUpstreamRequest(e.to_string()));
+      }
+      Ok(v) => v,
+    };
+    debug!(
+      "Request to be forwarded: uri {}, version {:?}, headers {:?}",
+      req.uri(),
+      req.version(),
+      req.headers()
+    );
+    log_data.xff(&req.headers().get("x-forwarded-for"));
+    log_data.upstream(req.uri());
+    //////
+
+    //////////////
+    // // TODO: remove later
+    let body = crate::hyper_ext::body::full(hyper::body::Bytes::from("not yet implemented"));
+    let mut res_backend = super::synthetic_response::synthetic_response(Response::builder().body(body).unwrap());
+    // // Forward request to a chosen backend
+    // let mut res_backend = {
+    //   let Ok(result) = timeout(self.globals.proxy_config.upstream_timeout, self.forwarder.request(req)).await else {
+    //     return self.return_with_error_log(StatusCode::GATEWAY_TIMEOUT, &mut log_data);
+    //   };
+    //   match result {
+    //     Ok(res) => res,
+    //     Err(e) => {
+    //       error!("Failed to get response from backend: {}", e);
+    //       return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data);
+    //     }
+    //   }
+    // };
+    //////////////
+    // Process reverse proxy context generated during the forwarding request generation.
+    #[cfg(feature = "sticky-cookie")]
+    if let Some(context_from_lb) = _context.context_lb {
+      let res_headers = res_backend.headers_mut();
+      if let Err(e) = set_sticky_cookie_lb_context(res_headers, &context_from_lb) {
+        error!("Failed to append context to the response given from backend: {}", e);
+        return Err(HttpError::FailedToAddSetCookeInResponse);
+      }
+    }
+
+    if res_backend.status() != StatusCode::SWITCHING_PROTOCOLS {
+      // Generate response to client
+      if let Err(e) = self.generate_response_forwarded(&mut res_backend, backend_app) {
+        error!("Failed to generate downstream response for clients: {}", e);
+        return Err(HttpError::FailedToGenerateDownstreamResponse(e.to_string()));
+      }
+      return Ok(res_backend);
+    }
+
+    // Handle StatusCode::SWITCHING_PROTOCOLS in response
+    let upgrade_in_response = extract_upgrade(res_backend.headers());
+    let should_upgrade = match (upgrade_in_request.as_ref(), upgrade_in_response.as_ref()) {
+      (Some(u_req), Some(u_res)) => u_req.to_ascii_lowercase() == u_res.to_ascii_lowercase(),
+      _ => false,
+    };
+
+    if !should_upgrade {
+      error!(
+        "Backend tried to switch to protocol {:?} when {:?} was requested",
+        upgrade_in_response, upgrade_in_request
+      );
+      return Err(HttpError::FailedToUpgrade);
+    }
+    let Some(request_upgraded) = request_upgraded else {
+      error!("Request does not have an upgrade extension");
+      return Err(HttpError::NoUpgradeExtensionInRequest);
+    };
+    let Some(onupgrade) = res_backend.extensions_mut().remove::<hyper::upgrade::OnUpgrade>() else {
+      error!("Response does not have an upgrade extension");
+      return Err(HttpError::NoUpgradeExtensionInResponse);
+    };
+
+    self.globals.runtime_handle.spawn(async move {
+      let mut response_upgraded = TokioIo::new(onupgrade.await.map_err(|e| {
+        error!("Failed to upgrade response: {}", e);
+        RpxyError::FailedToUpgradeResponse(e.to_string())
+      })?);
+      let mut request_upgraded = TokioIo::new(request_upgraded.await.map_err(|e| {
+        error!("Failed to upgrade request: {}", e);
+        RpxyError::FailedToUpgradeRequest(e.to_string())
+      })?);
+      copy_bidirectional(&mut response_upgraded, &mut request_upgraded)
+        .await
+        .map_err(|e| {
+          error!("Coping between upgraded connections failed: {}", e);
+          RpxyError::FailedToCopyBidirectional(e.to_string())
+        })?;
+      Ok(()) as RpxyResult<()>
+    });
+
+    Ok(res_backend)
+  }
+}