Merge pull request #76 from junkurihara/feat/http-version

feat: changed options for http version of requests to upstream app servers

TODO.md

@@ -1,5 +1,6 @@
 # TODO List
 
+- [Try in v0.5.1 or 0.6.0] Fix strategy for `h2c` requests on forwarded requests upstream. This needs to update forwarder definition. Also, maybe forwarder would have a cache corresponding to the following task.
 - [Try in v0.6.0] **Cache option for the response with `Cache-Control: public` header directive ([#55](https://github.com/junkurihara/rust-rpxy/issues/55))**
 - Improvement of path matcher
 - More flexible option for rewriting path

config-example.toml

@@ -56,7 +56,10 @@ upstream = [
   { location = 'www.yahoo.co.jp', tls = true },
 ]
 load_balance = "round_robin" # or "random" or "sticky" (sticky session) or "none" (fix to the first one, default)
-upstream_options = ["override_host", "convert_https_to_2"]
+upstream_options = [
+  "override_host",
+  "force_http2_upstream", # mutually exclusive with "force_http11_upstream"
+]
 
 # Non-default destination in "localhost" app, which is routed by "path"
 [[apps.localhost.reverse_proxy]]
@@ -75,7 +78,7 @@ load_balance = "random" # or "round_robin" or "sticky" (sticky session) or "none
 upstream_options = [
   "override_host",
   "upgrade_insecure_requests",
-  "convert_https_to_11",
+  "force_http11_upstream",
 ]
 ######################################################################
 

quinn

@@ -1 +1 @@
-Subproject commit 70e14b5c26b45ee1e3d5dd64b2a184e2d6376880
+Subproject commit 8076ffe94d38813ce0220af9d3438e7bfb5e8429

rpxy-lib/src/backend/upstream_opts.rs

@@ -4,8 +4,8 @@ use crate::error::*;
 pub enum UpstreamOption {
   OverrideHost,
   UpgradeInsecureRequests,
-  ConvertHttpsTo11,
-  ConvertHttpsTo2,
+  ForceHttp11Upstream,
+  ForceHttp2Upstream,
   // TODO: Adds more options for heder override
 }
 impl TryFrom<&str> for UpstreamOption {
@@ -14,8 +14,8 @@ impl TryFrom<&str> for UpstreamOption {
     match val {
       "override_host" => Ok(Self::OverrideHost),
       "upgrade_insecure_requests" => Ok(Self::UpgradeInsecureRequests),
-      "convert_https_to_11" => Ok(Self::ConvertHttpsTo11),
-      "convert_https_to_2" => Ok(Self::ConvertHttpsTo2),
+      "force_http11_upstream" => Ok(Self::ForceHttp11Upstream),
+      "force_http2_upstream" => Ok(Self::ForceHttp2Upstream),
       _ => Err(RpxyError::Other(anyhow!("Unsupported header option"))),
     }
   }

rpxy-lib/src/error.rs

@@ -7,13 +7,13 @@ pub type Result<T> = std::result::Result<T, RpxyError>;
 /// Describes things that can go wrong in the Rpxy
 #[derive(Debug, Error)]
 pub enum RpxyError {
-  #[error("Proxy build error")]
+  #[error("Proxy build error: {0}")]
   ProxyBuild(#[from] crate::proxy::ProxyBuilderError),
 
-  #[error("Backend build error")]
+  #[error("Backend build error: {0}")]
   BackendBuild(#[from] crate::backend::BackendBuilderError),
 
-  #[error("MessageHandler build error")]
+  #[error("MessageHandler build error: {0}")]
   HandlerBuild(#[from] crate::handler::HttpMessageHandlerBuilderError),
 
   #[error("Config builder error: {0}")]
@@ -32,40 +32,40 @@ pub enum RpxyError {
   #[error("LoadBalance Layer Error: {0}")]
   LoadBalance(String),
 
-  #[error("I/O Error")]
+  #[error("I/O Error: {0}")]
   Io(#[from] io::Error),
 
   // #[error("Toml Deserialization Error")]
   // TomlDe(#[from] toml::de::Error),
   #[cfg(feature = "http3-quinn")]
-  #[error("Quic Connection Error")]
+  #[error("Quic Connection Error [quinn]: {0}")]
   QuicConn(#[from] quinn::ConnectionError),
 
   #[cfg(feature = "http3-s2n")]
-  #[error("Quic Connection Error [s2n-quic]")]
+  #[error("Quic Connection Error [s2n-quic]: {0}")]
   QUicConn(#[from] s2n_quic::connection::Error),
 
   #[cfg(feature = "http3-quinn")]
-  #[error("H3 Error")]
+  #[error("H3 Error [quinn]: {0}")]
   H3(#[from] h3::Error),
 
   #[cfg(feature = "http3-s2n")]
-  #[error("H3 Error [s2n-quic]")]
+  #[error("H3 Error [s2n-quic]: {0}")]
   H3(#[from] s2n_quic_h3::h3::Error),
 
-  #[error("rustls Connection Error")]
+  #[error("rustls Connection Error: {0}")]
   Rustls(#[from] rustls::Error),
 
-  #[error("Hyper Error")]
+  #[error("Hyper Error: {0}")]
   Hyper(#[from] hyper::Error),
 
-  #[error("Hyper Http Error")]
+  #[error("Hyper Http Error: {0}")]
   HyperHttp(#[from] hyper::http::Error),
 
-  #[error("Hyper Http HeaderValue Error")]
+  #[error("Hyper Http HeaderValue Error: {0}")]
   HyperHeaderValue(#[from] hyper::header::InvalidHeaderValue),
 
-  #[error("Hyper Http HeaderName Error")]
+  #[error("Hyper Http HeaderName Error: {0}")]
   HyperHeaderName(#[from] hyper::header::InvalidHeaderName),
 
   #[error(transparent)]

rpxy-lib/src/globals.rs

@@ -225,7 +225,8 @@ where
     }
 
     if !(upstream.iter().all(|(_, elem)| {
-      !(elem.opts.contains(&UpstreamOption::ConvertHttpsTo11) && elem.opts.contains(&UpstreamOption::ConvertHttpsTo2))
+      !(elem.opts.contains(&UpstreamOption::ForceHttp11Upstream)
+        && elem.opts.contains(&UpstreamOption::ForceHttp2Upstream))
     })) {
       error!("Either one of force_http11 or force_http2 can be enabled");
       return Err(RpxyError::ConfigBuild("Invalid upstream option setting"));

rpxy-lib/src/handler/forwarder.rs

@@ -0,0 +1,69 @@
+use crate::error::RpxyError;
+use async_trait::async_trait;
+use derive_builder::Builder;
+use hyper::{
+  body::{Body, HttpBody},
+  client::{connect::Connect, HttpConnector},
+  http::Version,
+  Client, Request, Response,
+};
+use hyper_rustls::HttpsConnector;
+
+#[async_trait]
+/// Definition of the forwarder that simply forward requests from downstream client to upstream app servers.
+pub trait ForwardRequest<B> {
+  type Error;
+  async fn request(&self, req: Request<B>) -> Result<Response<Body>, Self::Error>;
+}
+
+#[derive(Builder, Clone)]
+/// Forwarder struct
+pub struct Forwarder<C, B = Body>
+where
+  C: Connect + Clone + Sync + Send + 'static,
+{
+  // TODO: maybe this forwarder definition is suitable for cache handling.
+  inner: Client<C, B>,
+  inner_h2: Client<C, B>, // `h2c` or http/2-only client is defined separately
+}
+
+#[async_trait]
+impl<C, B> ForwardRequest<B> for Forwarder<C, B>
+where
+  B: HttpBody + Send + Sync + 'static,
+  B::Data: Send,
+  B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
+  C: Connect + Clone + Sync + Send + 'static,
+{
+  type Error = RpxyError;
+  async fn request(&self, req: Request<B>) -> Result<Response<Body>, Self::Error> {
+    // TODO: This 'match' condition is always evaluated at every 'request' invocation. So, it is inefficient.
+    // Needs to be reconsidered. Currently, this is a kind of work around.
+    // This possibly relates to https://github.com/hyperium/hyper/issues/2417.
+    match req.version() {
+      Version::HTTP_2 => self.inner_h2.request(req).await.map_err(RpxyError::Hyper), // handles `h2c` requests
+      _ => self.inner.request(req).await.map_err(RpxyError::Hyper),
+    }
+  }
+}
+
+impl Forwarder<HttpsConnector<HttpConnector>, Body> {
+  pub async fn new() -> Self {
+    // let connector = TrustDnsResolver::default().into_rustls_webpki_https_connector();
+    let connector = hyper_rustls::HttpsConnectorBuilder::new()
+      .with_webpki_roots()
+      .https_or_http()
+      .enable_http1()
+      .enable_http2()
+      .build();
+    let connector_h2 = hyper_rustls::HttpsConnectorBuilder::new()
+      .with_webpki_roots()
+      .https_or_http()
+      .enable_http1()
+      .build();
+
+    let inner = Client::builder().build::<_, Body>(connector);
+    let inner_h2 = Client::builder().http2_only(true).build::<_, Body>(connector_h2);
+    Self { inner, inner_h2 }
+  }
+}

rpxy-lib/src/handler/handler_main.rs

@@ -1,5 +1,11 @@
 // Highly motivated by https://github.com/felipenoris/hyper-reverse-proxy
-use super::{utils_headers::*, utils_request::*, utils_synth_response::*, HandlerContext};
+use super::{
+  forwarder::{ForwardRequest, Forwarder},
+  utils_headers::*,
+  utils_request::*,
+  utils_synth_response::*,
+  HandlerContext,
+};
 use crate::{
   backend::{Backend, UpstreamGroup},
   certs::CryptoSource,
@@ -14,7 +20,7 @@ use hyper::{
   client::connect::Connect,
   header::{self, HeaderValue},
   http::uri::Scheme,
-  Body, Client, Request, Response, StatusCode, Uri, Version,
+  Body, Request, Response, StatusCode, Uri, Version,
 };
 use std::{net::SocketAddr, sync::Arc};
 use tokio::{io::copy_bidirectional, time::timeout};
@@ -27,7 +33,7 @@ where
   T: Connect + Clone + Sync + Send + 'static,
   U: CryptoSource + Clone,
 {
-  forwarder: Arc<Client<T>>,
+  forwarder: Arc<Forwarder<T>>,
   globals: Arc<Globals<U>>,
 }
 
@@ -357,14 +363,20 @@ where
     }
 
     // If not specified (force_httpXX_upstream) and https, version is preserved except for http/3
-    apply_upstream_options_to_request_line(req, upstream_group)?;
-    // Maybe workaround: Change version to http/1.1 when destination scheme is http
-    if req.version() != Version::HTTP_11 && upstream_chosen.uri.scheme() == Some(&Scheme::HTTP) {
-      *req.version_mut() = Version::HTTP_11;
-    } else if req.version() == Version::HTTP_3 {
-      debug!("HTTP/3 is currently unsupported for request to upstream. Use HTTP/2.");
+    match req.version() {
+      Version::HTTP_3 => {
+        debug!("HTTP/3 is currently unsupported for request to upstream.");
         *req.version_mut() = Version::HTTP_2;
       }
+      _ => {
+        if upstream_chosen.uri.scheme() == Some(&Scheme::HTTP) {
+          // Change version to http/1.1 when destination scheme is http
+          debug!("Change version to http/1.1 when destination scheme is http.");
+          *req.version_mut() = Version::HTTP_11;
+        }
+      }
+    }
+    apply_upstream_options_to_request_line(req, upstream_group)?;
 
     Ok(context)
   }

rpxy-lib/src/handler/mod.rs

@@ -1,3 +1,4 @@
+mod forwarder;
 mod handler_main;
 mod utils_headers;
 mod utils_request;
@@ -5,7 +6,10 @@ mod utils_synth_response;
 
 #[cfg(feature = "sticky-cookie")]
 use crate::backend::LbContext;
-pub use handler_main::{HttpMessageHandler, HttpMessageHandlerBuilder, HttpMessageHandlerBuilderError};
+pub use {
+  forwarder::Forwarder,
+  handler_main::{HttpMessageHandler, HttpMessageHandlerBuilder, HttpMessageHandlerBuilderError},
+};
 
 #[allow(dead_code)]
 #[derive(Debug)]

rpxy-lib/src/handler/utils_headers.rs

@@ -23,7 +23,7 @@ pub(super) fn takeout_sticky_cookie_lb_context(
 ) -> Result<Option<LbContext>> {
   let mut headers_clone = headers.clone();
 
-  match headers_clone.entry(hyper::header::COOKIE) {
+  match headers_clone.entry(header::COOKIE) {
     header::Entry::Vacant(_) => Ok(None),
     header::Entry::Occupied(entry) => {
       let cookies_iter = entry
@@ -43,8 +43,8 @@ pub(super) fn takeout_sticky_cookie_lb_context(
       }
       let cookies_passed_to_upstream = without_sticky_cookies.join("; ");
       let cookie_passed_to_lb = sticky_cookies.first().unwrap();
-      headers.remove(hyper::header::COOKIE);
-      headers.insert(hyper::header::COOKIE, cookies_passed_to_upstream.parse()?);
+      headers.remove(header::COOKIE);
+      headers.insert(header::COOKIE, cookies_passed_to_upstream.parse()?);
 
       let sticky_cookie = StickyCookie {
         value: StickyCookieValue::try_from(cookie_passed_to_lb, expected_cookie_name)?,
@@ -63,7 +63,7 @@ pub(super) fn set_sticky_cookie_lb_context(headers: &mut HeaderMap, context_from
   let sticky_cookie_string: String = context_from_lb.sticky_cookie.clone().try_into()?;
   let new_header_val: HeaderValue = sticky_cookie_string.parse()?;
   let expected_cookie_name = &context_from_lb.sticky_cookie.value.name;
-  match headers.entry(hyper::header::SET_COOKIE) {
+  match headers.entry(header::SET_COOKIE) {
     header::Entry::Vacant(entry) => {
       entry.insert(new_header_val);
     }
@@ -173,13 +173,13 @@ pub(super) fn add_header_entry_overwrite_if_exist(
 pub(super) fn make_cookie_single_line(headers: &mut HeaderMap) -> Result<()> {
   let cookies = headers
     .iter()
-    .filter(|(k, _)| **k == hyper::header::COOKIE)
+    .filter(|(k, _)| **k == header::COOKIE)
     .map(|(_, v)| v.to_str().unwrap_or(""))
     .collect::<Vec<_>>()
     .join("; ");
   if !cookies.is_empty() {
-    headers.remove(hyper::header::COOKIE);
-    headers.insert(hyper::header::COOKIE, HeaderValue::from_bytes(cookies.as_bytes())?);
+    headers.remove(header::COOKIE);
+    headers.insert(header::COOKIE, HeaderValue::from_bytes(cookies.as_bytes())?);
   }
   Ok(())
 }

rpxy-lib/src/handler/utils_request.rs

@@ -11,8 +11,12 @@ use hyper::{header, Request};
 pub(super) fn apply_upstream_options_to_request_line<B>(req: &mut Request<B>, upstream: &UpstreamGroup) -> Result<()> {
   for opt in upstream.opts.iter() {
     match opt {
-      UpstreamOption::ConvertHttpsTo11 => *req.version_mut() = hyper::Version::HTTP_11,
-      UpstreamOption::ConvertHttpsTo2 => *req.version_mut() = hyper::Version::HTTP_2,
+      UpstreamOption::ForceHttp11Upstream => *req.version_mut() = hyper::Version::HTTP_11,
+      UpstreamOption::ForceHttp2Upstream => {
+        // case: h2c -> https://www.rfc-editor.org/rfc/rfc9113.txt
+        // Upgrade from HTTP/1.1 to HTTP/2 is deprecated. So, http-2 prior knowledge is required.
+        *req.version_mut() = hyper::Version::HTTP_2;
+      }
       _ => (),
     }
   }

rpxy-lib/src/lib.rs

@@ -8,9 +8,14 @@ mod log;
 mod proxy;
 mod utils;
 
-use crate::{error::*, globals::Globals, handler::HttpMessageHandlerBuilder, log::*, proxy::ProxyBuilder};
+use crate::{
+  error::*,
+  globals::Globals,
+  handler::{Forwarder, HttpMessageHandlerBuilder},
+  log::*,
+  proxy::ProxyBuilder,
+};
 use futures::future::select_all;
-use hyper::Client;
 // use hyper_trust_dns::TrustDnsResolver;
 use std::sync::Arc;
 
@@ -62,16 +67,10 @@ where
     request_count: Default::default(),
     runtime_handle: runtime_handle.clone(),
   });
-  // let connector = TrustDnsResolver::default().into_rustls_webpki_https_connector();
-  let connector = hyper_rustls::HttpsConnectorBuilder::new()
-    .with_webpki_roots()
-    .https_or_http()
-    .enable_http1()
-    .enable_http2()
-    .build();
 
+  // TODO: HTTP2 only client is needed for http2 cleartext case
   let msg_handler = HttpMessageHandlerBuilder::default()
-    .forwarder(Arc::new(Client::builder().build::<_, hyper::Body>(connector)))
+    .forwarder(Arc::new(Forwarder::new().await))
     .globals(globals.clone())
     .build()?;
 

rpxy-lib/src/log.rs

@@ -1,4 +1,5 @@
 use crate::utils::ToCanonical;
+use hyper::header;
 use std::net::SocketAddr;
 pub use tracing::{debug, error, info, warn};
 
@@ -20,7 +21,7 @@ pub struct MessageLog {
 
 impl<T> From<&hyper::Request<T>> for MessageLog {
   fn from(req: &hyper::Request<T>) -> Self {
-    let header_mapper = |v: hyper::header::HeaderName| {
+    let header_mapper = |v: header::HeaderName| {
       req
         .headers()
         .get(v)
@@ -31,7 +32,7 @@ impl<T> From<&hyper::Request<T>> for MessageLog {
       // tls_server_name: "".to_string(),
       client_addr: "".to_string(),
       method: req.method().to_string(),
-      host: header_mapper(hyper::header::HOST),
+      host: header_mapper(header::HOST),
       p_and_q: req
         .uri()
         .path_and_query()
@@ -40,8 +41,8 @@ impl<T> From<&hyper::Request<T>> for MessageLog {
       version: req.version(),
       uri_scheme: req.uri().scheme_str().unwrap_or("").to_string(),
       uri_host: req.uri().host().unwrap_or("").to_string(),
-      ua: header_mapper(hyper::header::USER_AGENT),
-      xff: header_mapper(hyper::header::HeaderName::from_static("x-forwarded-for")),
+      ua: header_mapper(header::USER_AGENT),
+      xff: header_mapper(header::HeaderName::from_static("x-forwarded-for")),
       status: "".to_string(),
       upstream: "".to_string(),
     }
@@ -61,7 +62,7 @@ impl MessageLog {
     self.status = status_code.to_string();
     self
   }
-  pub fn xff(&mut self, xff: &Option<&hyper::header::HeaderValue>) -> &mut Self {
+  pub fn xff(&mut self, xff: &Option<&header::HeaderValue>) -> &mut Self {
     self.xff = xff.map_or_else(|| "", |v| v.to_str().unwrap_or("")).to_string();
     self
   }

s2n-quic

@@ -1 +1 @@
-Subproject commit 8ef0a6b66a856dc9f34ce18159c617ac29154cc7
+Subproject commit 1ff2cd230fdf46596fe77830966857c438a8b31a