Merge pull request #307 from junkurihara/feat/forwarded-header

Feat: forwarded header

.gitignore

@@ -1,5 +1,6 @@
 .vscode
 .private
+.tmp
 docker/log
 docker/cache
 docker/config

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## 0.10.1 or 0.11.0 (Unreleased)
 
+### Improvement
+
+- Feat: Support `Forwarded` header in addition to `X-Forwarded-For` header. This is to support the standard forwarding header for reverse proxy applications (RFC 7239). Use the `forwarded_header` upstream option to enable this feature.
+  By default, it is not appended to the outgoing header. However, if the incoming request has the forwarded header, it would be preserved and updated simultaneously with `x-forwarded-for` header. if both forwarded and x-forwarded-for headers exists (and they are inconsistent), x-forwarded-for is prioritized. This means that x-forwarded-for is first updated and it is then copied (overridden) to `for` param of forwarded header.
+
 ## 0.10.0
 
 ### Important Changes

config-example.toml

@@ -84,6 +84,7 @@ upstream_options = [
   "upgrade_insecure_requests",
   "force_http11_upstream",
   "set_upstream_host",        # overwrite HOST value with upstream hostname (like www.yahoo.com)
+  "forwarded_header"          # add Forwarded header (by default, this is not added. However, if the incoming request has Forwarded header, it would be preserved and updated)
 ]
 ######################################################################
 

rpxy-lib/src/backend/upstream_opts.rs

@@ -13,6 +13,8 @@ pub enum UpstreamOption {
   ForceHttp11Upstream,
   /// Force HTTP/2 upstream
   ForceHttp2Upstream,
+  /// Add RFC 7239 Forwarded header
+  ForwardedHeader,
   // TODO: Adds more options for heder override
 }
 impl TryFrom<&str> for UpstreamOption {
@@ -24,6 +26,7 @@ impl TryFrom<&str> for UpstreamOption {
       "upgrade_insecure_requests" => Ok(Self::UpgradeInsecureRequests),
       "force_http11_upstream" => Ok(Self::ForceHttp11Upstream),
       "force_http2_upstream" => Ok(Self::ForceHttp2Upstream),
+      "forwarded_header" => Ok(Self::ForwardedHeader),
       _ => Err(RpxyError::UnsupportedUpstreamOption),
     }
   }

rpxy-lib/src/message_handler/handler_manipulate_messages.rs

@@ -81,13 +81,13 @@ where
         .unwrap_or(false)
     };
 
-    let original_uri = req.uri().to_string();
+    let original_uri = req.uri().clone();
     let headers = req.headers_mut();
     // delete headers specified in header.connection
     remove_connection_header(headers);
     // delete hop headers including header.connection
     remove_hop_header(headers);
-    // X-Forwarded-For
+    // X-Forwarded-For (and Forwarded if exists)
     add_forwarding_header(headers, client_addr, listen_addr, tls_enabled, &original_uri)?;
 
     // Add te: trailer if te_trailer
@@ -126,8 +126,8 @@ where
 
     // apply upstream-specific headers given in upstream_option
     let headers = req.headers_mut();
-    // apply upstream options to header
-    apply_upstream_options_to_header(headers, &upstream_chosen.uri, upstream_candidates)?;
+    // apply upstream options to header, after X-Forwarded-For is added
+    apply_upstream_options_to_header(headers, &upstream_chosen.uri, upstream_candidates, &original_uri)?;
 
     // update uri in request
     ensure!(

rpxy-lib/src/message_handler/http_log.rs

@@ -1,5 +1,5 @@
 use super::canonical_address::ToCanonical;
-use crate::log::*;
+use crate::{log::*, message_handler::utils_headers};
 use http::header;
 use std::net::SocketAddr;
 
@@ -12,10 +12,11 @@ pub struct HttpMessageLog {
   pub host: String,
   pub p_and_q: String,
   pub version: http::Version,
-  pub uri_scheme: String,
-  pub uri_host: String,
+  pub scheme: String,
+  pub path: String,
   pub ua: String,
   pub xff: String,
+  pub forwarded: String,
   pub status: String,
   pub upstream: String,
 }
@@ -29,17 +30,21 @@ impl<T> From<&http::Request<T>> for HttpMessageLog {
         .map_or_else(|| "", |s| s.to_str().unwrap_or(""))
         .to_string()
     };
+    let host =
+      utils_headers::host_from_uri_or_host_header(req.uri(), req.headers().get(header::HOST).cloned()).unwrap_or_default();
+
     Self {
       // tls_server_name: "".to_string(),
       client_addr: "".to_string(),
       method: req.method().to_string(),
-      host: header_mapper(header::HOST),
+      host,
       p_and_q: req.uri().path_and_query().map_or_else(|| "", |v| v.as_str()).to_string(),
       version: req.version(),
-      uri_scheme: req.uri().scheme_str().unwrap_or("").to_string(),
-      uri_host: req.uri().host().unwrap_or("").to_string(),
+      scheme: req.uri().scheme_str().unwrap_or("").to_string(),
+      path: req.uri().path().to_string(),
       ua: header_mapper(header::USER_AGENT),
       xff: header_mapper(header::HeaderName::from_static("x-forwarded-for")),
+      forwarded: header_mapper(header::FORWARDED),
       status: "".to_string(),
       upstream: "".to_string(),
     }
@@ -48,26 +53,29 @@ impl<T> From<&http::Request<T>> for HttpMessageLog {
 
 impl std::fmt::Display for HttpMessageLog {
   fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    let forwarded_part = if !self.forwarded.is_empty() {
+      format!(" \"{}\"", self.forwarded)
+    } else {
+      "".to_string()
+    };
+
     write!(
       f,
-      "{} <- {} -- {} {} {:?} -- {} -- {} \"{}\", \"{}\" \"{}\"",
-      if !self.host.is_empty() {
-        self.host.as_str()
-      } else {
-        self.uri_host.as_str()
-      },
+      "{} <- {} -- {} {} {:?} -- {} -- {} \"{}\", \"{}\"{} \"{}\"",
+      self.host,
       self.client_addr,
       self.method,
       self.p_and_q,
       self.version,
       self.status,
-      if !self.uri_scheme.is_empty() && !self.uri_host.is_empty() {
-        format!("{}://{}", self.uri_scheme, self.uri_host)
+      if !self.scheme.is_empty() && !self.host.is_empty() {
+        format!("{}://{}{}", self.scheme, self.host, self.path)
       } else {
-        "".to_string()
+        self.path.clone()
       },
       self.ua,
       self.xff,
+      forwarded_part,
       self.upstream
     )
   }
@@ -102,3 +110,56 @@ impl HttpMessageLog {
     );
   }
 }
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use http::{Method, Version};
+
+  #[test]
+  fn test_log_format_without_forwarded() {
+    let log = HttpMessageLog {
+      client_addr: "192.168.1.1:8080".to_string(),
+      method: Method::GET.to_string(),
+      host: "example.com".to_string(),
+      p_and_q: "/path?query=value".to_string(),
+      version: Version::HTTP_11,
+      scheme: "https".to_string(),
+      path: "/path".to_string(),
+      ua: "Mozilla/5.0".to_string(),
+      xff: "10.0.0.1".to_string(),
+      forwarded: "".to_string(),
+      status: "200".to_string(),
+      upstream: "https://backend.example.com".to_string(),
+    };
+
+    let formatted = format!("{}", log);
+    assert!(!formatted.contains(" \"\""));
+    assert!(formatted.contains("\"Mozilla/5.0\", \"10.0.0.1\" \"https://backend.example.com\""));
+  }
+
+  #[test]
+  fn test_log_format_with_forwarded() {
+    let log = HttpMessageLog {
+      client_addr: "192.168.1.1:8080".to_string(),
+      method: Method::GET.to_string(),
+      host: "example.com".to_string(),
+      p_and_q: "/path?query=value".to_string(),
+      version: Version::HTTP_11,
+      scheme: "https".to_string(),
+      path: "/path".to_string(),
+      ua: "Mozilla/5.0".to_string(),
+      xff: "10.0.0.1".to_string(),
+      forwarded: "for=192.0.2.60;proto=http;by=203.0.113.43".to_string(),
+      status: "200".to_string(),
+      upstream: "https://backend.example.com".to_string(),
+    };
+
+    let formatted = format!("{}", log);
+    assert!(formatted.contains(" \"for=192.0.2.60;proto=http;by=203.0.113.43\""));
+    assert!(
+      formatted
+        .contains("\"Mozilla/5.0\", \"10.0.0.1\" \"for=192.0.2.60;proto=http;by=203.0.113.43\" \"https://backend.example.com\"")
+    );
+  }
+}

rpxy-lib/src/message_handler/utils_headers.rs

@@ -12,6 +12,13 @@ use std::{borrow::Cow, net::SocketAddr};
 use crate::backend::{LoadBalanceContext, StickyCookie, StickyCookieValue};
 // use crate::backend::{UpstreamGroup, UpstreamOption};
 
+const X_FORWARDED_FOR: &str = "x-forwarded-for";
+const X_FORWARDED_PROTO: &str = "x-forwarded-proto";
+const X_FORWARDED_PORT: &str = "x-forwarded-port";
+const X_FORWARDED_SSL: &str = "x-forwarded-ssl";
+const X_ORIGINAL_URI: &str = "x-original-uri";
+const X_REAL_IP: &str = "x-real-ip";
+
 // ////////////////////////////////////////////////////
 // // Functions to manipulate headers
 #[cfg(feature = "sticky-cookie")]
@@ -96,11 +103,13 @@ fn override_host_header(headers: &mut HeaderMap, upstream_base_uri: &Uri) -> Res
 }
 
 /// Apply options to request header, which are specified in the configuration
+/// This function is called after almost all other headers has been set and updated.
 pub(super) fn apply_upstream_options_to_header(
   headers: &mut HeaderMap,
   upstream_base_uri: &Uri,
   // _client_addr: &SocketAddr,
   upstream: &UpstreamCandidates,
+  original_uri: &Uri,
 ) -> Result<()> {
   for opt in upstream.options.iter() {
     match opt {
@@ -117,6 +126,21 @@ pub(super) fn apply_upstream_options_to_header(
           .entry(header::UPGRADE_INSECURE_REQUESTS)
           .or_insert(HeaderValue::from_bytes(b"1").unwrap());
       }
+      UpstreamOption::ForwardedHeader => {
+        // This is called after X-Forwarded-For is added
+        // Generate RFC 7239 Forwarded header
+        let tls = upstream_base_uri.scheme_str() == Some("https");
+
+        match generate_forwarded_header(headers, tls, original_uri) {
+          Ok(forwarded_value) => {
+            add_header_entry_overwrite_if_exist(headers, header::FORWARDED.as_str(), forwarded_value)?;
+          }
+          Err(e) => {
+            // Log warning but don't fail the request if Forwarded generation fails
+            warn!("Failed to generate Forwarded header: {}", e);
+          }
+        }
+      }
       _ => (),
     }
   }
@@ -194,18 +218,45 @@ pub(super) fn make_cookie_single_line(headers: &mut HeaderMap) -> Result<()> {
   Ok(())
 }
 
-/// Add forwarding headers like `x-forwarded-for`.
+/// Add or update forwarding headers like `x-forwarded-for`.
+/// If only `forwarded` header exists, it will update `x-forwarded-for` with the proxy chain.
+/// If both `x-forwarded-for` and `forwarded` headers exist, it will update `x-forwarded-for` first and then add `forwarded` header.
 pub(super) fn add_forwarding_header(
   headers: &mut HeaderMap,
   client_addr: &SocketAddr,
   listen_addr: &SocketAddr,
   tls: bool,
-  uri_str: &str,
+  original_uri: &Uri,
 ) -> Result<()> {
-  // default process
-  // optional process defined by upstream_option is applied in fn apply_upstream_options
   let canonical_client_addr = client_addr.to_canonical().ip().to_string();
-  append_header_entry_with_comma(headers, "x-forwarded-for", &canonical_client_addr)?;
+  let has_forwarded = headers.contains_key(header::FORWARDED);
+  let has_xff = headers.contains_key(X_FORWARDED_FOR);
+
+  // Handle incoming Forwarded header (Case 2: only Forwarded exists)
+  if has_forwarded && !has_xff {
+    // Extract proxy chain from Forwarded header and update X-Forwarded-For for consistency
+    update_xff_from_forwarded(headers, client_addr)?;
+  } else {
+    // Case 1: only X-Forwarded-For exists, or Case 3: both exist (conservative: use X-Forwarded-For)
+    // TODO: In future PR, implement proper RFC 7239 precedence
+    // where Forwarded header should take priority over X-Forwarded-For
+    // This requires careful testing to ensure no breaking changes
+    append_header_entry_with_comma(headers, X_FORWARDED_FOR, &canonical_client_addr)?;
+  }
+
+  // IMPORTANT: If Forwarded header exists, always update it for consistency
+  // This ensures headers remain consistent even when forwarded_header upstream option is not specified
+  if has_forwarded {
+    match generate_forwarded_header(headers, tls, original_uri) {
+      Ok(forwarded_value) => {
+        add_header_entry_overwrite_if_exist(headers, header::FORWARDED.as_str(), forwarded_value)?;
+      }
+      Err(e) => {
+        // Log warning but don't fail the request if Forwarded generation fails
+        warn!("Failed to update existing Forwarded header for consistency: {}", e);
+      }
+    }
+  }
 
   // Single line cookie header
   // TODO: This should be only for HTTP/1.1. For 2+, this can be multi-lined.
@@ -214,24 +265,127 @@ pub(super) fn add_forwarding_header(
   /////////// As Nginx
   // If we receive X-Forwarded-Proto, pass it through; otherwise, pass along the
   // scheme used to connect to this server
-  add_header_entry_if_not_exist(headers, "x-forwarded-proto", if tls { "https" } else { "http" })?;
+  add_header_entry_if_not_exist(headers, X_FORWARDED_PROTO, if tls { "https" } else { "http" })?;
   // If we receive X-Forwarded-Port, pass it through; otherwise, pass along the
   // server port the client connected to
-  add_header_entry_if_not_exist(headers, "x-forwarded-port", listen_addr.port().to_string())?;
+  add_header_entry_if_not_exist(headers, X_FORWARDED_PORT, listen_addr.port().to_string())?;
 
   /////////// As Nginx-Proxy
   // x-real-ip
-  add_header_entry_overwrite_if_exist(headers, "x-real-ip", canonical_client_addr)?;
+  add_header_entry_overwrite_if_exist(headers, X_REAL_IP, canonical_client_addr)?;
   // x-forwarded-ssl
-  add_header_entry_overwrite_if_exist(headers, "x-forwarded-ssl", if tls { "on" } else { "off" })?;
+  add_header_entry_overwrite_if_exist(headers, X_FORWARDED_SSL, if tls { "on" } else { "off" })?;
   // x-original-uri
-  add_header_entry_overwrite_if_exist(headers, "x-original-uri", uri_str.to_string())?;
+  add_header_entry_overwrite_if_exist(headers, X_ORIGINAL_URI, original_uri.to_string())?;
   // proxy
   add_header_entry_overwrite_if_exist(headers, "proxy", "")?;
 
   Ok(())
 }
 
+/// Extract proxy chain from existing Forwarded header
+fn extract_forwarded_chain(headers: &HeaderMap) -> Vec<String> {
+  headers
+    .get(header::FORWARDED)
+    .and_then(|h| h.to_str().ok())
+    .map(|forwarded_str| {
+      // Parse Forwarded header entries (comma-separated)
+      forwarded_str
+        .split(',')
+        .flat_map(|entry| entry.split(';'))
+        .map(str::trim)
+        .filter_map(|param| param.strip_prefix("for="))
+        .map(|for_value| {
+          // Remove quotes from IPv6 addresses for consistency with X-Forwarded-For
+          if let Some(ipv6) = for_value.strip_prefix("\"[").and_then(|s| s.strip_suffix("]\"")) {
+            ipv6.to_string()
+          } else {
+            for_value.to_string()
+          }
+        })
+        .collect()
+    })
+    .unwrap_or_default()
+}
+
+/// Update X-Forwarded-For with proxy chain from Forwarded header for consistency
+fn update_xff_from_forwarded(headers: &mut HeaderMap, client_addr: &SocketAddr) -> Result<()> {
+  let forwarded_chain = extract_forwarded_chain(headers);
+
+  if !forwarded_chain.is_empty() {
+    // Replace X-Forwarded-For with the chain from Forwarded header
+    headers.remove(X_FORWARDED_FOR);
+    for ip in forwarded_chain {
+      append_header_entry_with_comma(headers, X_FORWARDED_FOR, &ip)?;
+    }
+  }
+
+  // Append current client IP (standard behavior)
+  let canonical_client_addr = client_addr.to_canonical().ip().to_string();
+  append_header_entry_with_comma(headers, X_FORWARDED_FOR, &canonical_client_addr)?;
+
+  Ok(())
+}
+
+/// Generate RFC 7239 Forwarded header from X-Forwarded-For
+/// This function assumes that the X-Forwarded-For header is present and well-formed.
+fn generate_forwarded_header(headers: &HeaderMap, tls: bool, original_uri: &Uri) -> Result<String> {
+  let for_values = headers
+    .get(X_FORWARDED_FOR)
+    .and_then(|h| h.to_str().ok())
+    .map(|xff_str| {
+      xff_str
+        .split(',')
+        .map(str::trim)
+        .filter(|ip| !ip.is_empty())
+        .map(|ip| {
+          // Format IP according to RFC 7239 (quote IPv6)
+          if ip.contains(':') {
+            format!("\"[{}]\"", ip)
+          } else {
+            ip.to_string()
+          }
+        })
+        .collect::<Vec<_>>()
+        .join(",for=")
+    })
+    .unwrap_or_default();
+
+  if for_values.is_empty() {
+    return Err(anyhow!("No X-Forwarded-For header found for Forwarded generation"));
+  }
+
+  // Build forwarded header value
+  let forwarded_value = format!(
+    "for={};proto={};host={}",
+    for_values,
+    if tls { "https" } else { "http" },
+    host_from_uri_or_host_header(original_uri, headers.get(header::HOST).cloned())?
+  );
+
+  Ok(forwarded_value)
+}
+
+/// Extract host from URI
+pub(super) fn host_from_uri_or_host_header(uri: &Uri, host_header_value: Option<header::HeaderValue>) -> Result<String> {
+  // Prioritize uri host over host header
+  let uri_host = uri.host().map(|host| {
+    if let Some(port) = uri.port_u16() {
+      format!("{}:{}", host, port)
+    } else {
+      host.to_string()
+    }
+  });
+  if let Some(host) = uri_host {
+    return Ok(host);
+  }
+  // If uri host is not available, use host header
+  host_header_value
+    .map(|h| h.to_str().map(|s| s.to_string()))
+    .transpose()?
+    .ok_or_else(|| anyhow!("No host found in URI or Host header"))
+}
+
 /// Remove connection header
 pub(super) fn remove_connection_header(headers: &mut HeaderMap) {
   if let Some(values) = headers.get(header::CONNECTION) {