v0.34.0_6

src/forwarding.rs

@@ -225,9 +225,10 @@ impl Compiled {
             let original_url = format!("{scheme}://{authority_text}{target}");
             let peer_text = peer_ip.to_string();
             let peer_header = HeaderValue::from_str(&peer_text)?;
+            let peer_node = if peer_ip.is_ipv4() { None } else { Some(node(peer_ip)) };
             let forwarded = format!(
                 "for={};by={};proto={scheme};host=\"{authority_text}\"",
-                node(peer_ip), node(canonical(local.ip()))
+                peer_node.as_deref().unwrap_or(&peer_text), node(canonical(local.ip()))
             );
             let mut values = HeaderMap::with_capacity(7);
             values.insert("forwarded", forwarded.parse()?);

src/proxy/body/tests.rs

@@ -46,6 +46,36 @@ mod tests {
         });
         (exchange, slots)
     }
+    #[tokio::test]
+    async fn optimized_empty_request_keeps_exchange_until_response() {
+        let (exchange, slots) = observation().await;
+        let metrics = exchange.metrics.clone();
+        let body = super::super::wrap_observed_request(
+            Body::empty(), &exchange, Duration::from_secs(1),
+        );
+        assert!(body.is_end_stream());
+        drop(body);
+        assert_eq!(slots.available_permits(), 0);
+        drop(exchange);
+        assert_eq!(slots.available_permits(), 1);
+        assert_eq!(metrics.completed.load(Ordering::Relaxed), 1);
+    }
+
+    #[tokio::test]
+    async fn optimized_nonempty_request_still_meters_data_and_holds_capacity() {
+        let (exchange, slots) = observation().await;
+        let metrics = exchange.metrics.clone();
+        let body = super::super::wrap_observed_request(
+            Body::from("payload"), &exchange, Duration::from_secs(1),
+        );
+        assert!(!body.is_end_stream());
+        drop(exchange);
+        assert_eq!(slots.available_permits(), 0);
+        assert_eq!(body.collect().await.unwrap().to_bytes().len(), 7);
+        assert_eq!(slots.available_permits(), 1);
+        assert_eq!(metrics.bytes_in.load(Ordering::Relaxed), 7);
+    }
+
     #[tokio::test]
     async fn permit_and_metrics_cover_both_bodies() {
         let (exchange, slots) = observation().await;

src/proxy/forward/cache_lookup.rs

@@ -40,7 +40,7 @@ if let Some(ticket) = &cache_ticket {
                         cache_settings: cache_settings.clone(),
                         transform: transform_settings.cloned(),
                         uri: uri.clone(),
-                        public_suffix: public_suffix.clone(),
+                        public_suffix: public_suffix.to_string(),
                         host: host.clone(),
                         path: path.as_ref().to_owned(),
                         scope: scope.to_owned(),
@@ -77,7 +77,7 @@ if let Some(ticket) = &cache_ticket {
                         cache_settings: cache_settings.clone(),
                         transform: transform_settings.cloned(),
                         uri: uri.clone(),
-                        public_suffix: public_suffix.clone(),
+                        public_suffix: public_suffix.to_string(),
                         host: host.clone(),
                         path: path.as_ref().to_owned(),
                         scope: scope.to_owned(),
@@ -238,7 +238,7 @@ if handshake.is_none()
                                     cache_settings: cache_settings.clone(),
                                     transform: transform_settings.cloned(),
                                     uri: uri.clone(),
-                                    public_suffix: public_suffix.clone(),
+                                    public_suffix: public_suffix.to_string(),
                                     host: host.clone(),
                                     path: path.as_ref().to_owned(),
                                     scope: scope.to_owned(),

src/proxy/forward/route_cache.rs

@@ -169,12 +169,11 @@ let body_empty = if cache_enabled
 } else {
     cache_enabled && request.body().is_end_stream()
 };
-let public_suffix = request
-    .uri()
-    .path_and_query()
-    .map(|v| v.as_str())
-    .unwrap_or("/")
-    .to_owned();
+// Keep the original URI alive through forwarding: ordinary targets can be
+// borrowed while route rewrites still allocate and are strictly validated.
+let public_suffix: std::borrow::Cow<'_, str> = std::borrow::Cow::Borrowed(
+    routing_uri.path_and_query().map(|v| v.as_str()).unwrap_or("/")
+);
 let cache_plan = cache_plan::CachePlan::new(policy_reason, cache_settings, &path, || {
     cache_settings.active_for(route.cache) && route
         .frontend
@@ -200,7 +199,7 @@ let origin_suffix = match route.rewritten_path(&path) {
             encoded.push('?');
             encoded.push_str(query);
         }
-        encoded
+        std::borrow::Cow::Owned(encoded)
     }
     Err(_) => return error(StatusCode::BAD_REQUEST, "RustProxy: invalid rewritten request target\n"),
 };

src/proxy/forward/success_finish.rs

@@ -168,7 +168,7 @@ if let (Some(ticket), Some(plan), Some(transformed)) =
                 cache_settings: cache_settings.clone(),
                 transform: transform_settings.cloned(),
                 uri: active_uri.clone(),
-                public_suffix: public_suffix.clone(),
+                public_suffix: public_suffix.to_string(),
                 host: host.clone(),
                 path: path.as_ref().to_owned(),
                 scope: scope.to_owned(),

src/proxy/forward/upstream.rs

@@ -92,7 +92,7 @@ let retry_host_follows_backend = replay_headers.as_ref()
     .and_then(|headers| headers.get(HOST))
     .is_some_and(|value| value == expected_backend_host);
 let Some(upstream_target) = upstream_request_target(
-    &uri, request.headers_mut(), &backend,
+    uri, request.headers_mut(), &backend,
     websocket_legacy_support && !websocket_upstream_h2, websocket_upstream_h2,
 ) else {
     return error(StatusCode::BAD_GATEWAY, "RustProxy: invalid upstream authority\n");

src/proxy/headers.rs

@@ -7,14 +7,17 @@ pub fn strip_hop_by_hop(headers: &mut HeaderMap) {
     // The common HTTP/1.1 "Connection: keep-alive" case has no dynamic
     // nominated headers to collect.  Keep the general multi-value/token
     // parser for all other requests (including maliciously repeated fields).
-    let only_keep_alive = {
+    let no_nominations = {
         let mut values = headers.get_all(CONNECTION).iter();
-        values.next().is_some_and(|value| {
-            value.as_bytes().eq_ignore_ascii_case(b"keep-alive")
-                && values.next().is_none()
-        })
+        match (values.next(), values.next()) {
+            (None, _) => true,
+            (Some(value), None) => value.as_bytes().eq_ignore_ascii_case(b"keep-alive")
+                || (value.as_bytes().eq_ignore_ascii_case(b"close")
+                    && !headers.contains_key("close")),
+            _ => false,
+        }
     };
-    if !only_keep_alive {
+    if !no_nominations {
         let nominated: Vec<HeaderName> = headers
             .get_all(CONNECTION)
             .iter()
@@ -41,13 +44,18 @@ pub fn strip_hop_by_hop(headers: &mut HeaderMap) {
     }
 }
 pub fn requests_upgrade(headers: &HeaderMap) -> bool {
-    headers.contains_key("upgrade")
-        || headers
-            .get_all(CONNECTION)
-            .iter()
-            .filter_map(|v| v.to_str().ok())
-            .flat_map(|v| v.split(','))
-            .any(|v| v.trim().eq_ignore_ascii_case("upgrade"))
+    if headers.contains_key("upgrade") { return true; }
+    let mut connection = headers.get_all(CONNECTION).iter();
+    match (connection.next(), connection.next()) {
+        (None, _) => return false,
+        (Some(value), None) if value.as_bytes().eq_ignore_ascii_case(b"keep-alive")
+            || value.as_bytes().eq_ignore_ascii_case(b"close") => return false,
+        _ => {}
+    }
+    headers.get_all(CONNECTION).iter()
+        .filter_map(|v| v.to_str().ok())
+        .flat_map(|v| v.split(','))
+        .any(|v| v.trim().eq_ignore_ascii_case("upgrade"))
 }
 #[cfg(test)]
 include!("headers/tests.rs");

src/proxy/headers/tests.rs

@@ -19,6 +19,21 @@ mod tests {
         h.insert("connection", "keep-alive, Upgrade".parse().unwrap());
         assert!(requests_upgrade(&h));
     }
+    #[test]
+    fn common_connection_tokens_do_not_request_websocket_upgrade() {
+        for connection in [None, Some("keep-alive"), Some("KEEP-ALIVE"), Some("close")] {
+            let mut headers = HeaderMap::new();
+            if let Some(value) = connection { headers.insert("connection", value.parse().unwrap()); }
+            assert!(!requests_upgrade(&headers));
+        }
+        let mut headers = HeaderMap::new();
+        headers.insert("connection", "close, Upgrade".parse().unwrap());
+        assert!(requests_upgrade(&headers));
+        headers.insert("connection", "keep-alive".parse().unwrap());
+        headers.insert("upgrade", "websocket".parse().unwrap());
+        assert!(requests_upgrade(&headers));
+    }
+
     #[test]
     fn absent_cookies_leave_headers_unchanged() {
         let mut headers = HeaderMap::new();
@@ -68,6 +83,31 @@ mod tests {
             assert_eq!(headers["authorization"], "Bearer allowed");
         }
     }
+    #[test]
+    fn absent_and_close_connection_preserve_non_hop_headers() {
+        for value in [None, Some("close"), Some("CLOSE")] {
+            let mut headers = HeaderMap::new();
+            if let Some(value) = value { headers.insert("connection", value.parse().unwrap()); }
+            headers.insert("x-request-policy", "allowed".parse().unwrap());
+            headers.insert("transfer-encoding", "chunked".parse().unwrap());
+            strip_hop_by_hop(&mut headers);
+            assert!(!headers.contains_key("connection"));
+            assert!(!headers.contains_key("transfer-encoding"));
+            assert_eq!(headers["x-request-policy"], "allowed");
+        }
+    }
+
+    #[test]
+    fn connection_close_still_strips_a_close_field_if_present() {
+        let mut headers = HeaderMap::new();
+        headers.insert("connection", "close".parse().unwrap());
+        headers.insert("close", "private-hop-value".parse().unwrap());
+        headers.insert("x-public", "allowed".parse().unwrap());
+        strip_hop_by_hop(&mut headers);
+        assert!(!headers.contains_key("close"));
+        assert_eq!(headers["x-public"], "allowed");
+    }
+
     #[test]
     fn multiple_connection_fields_still_strip_all_nominated_headers() {
         let mut headers = HeaderMap::new();

src/proxy/tests/authority_tests.rs

@@ -58,7 +58,7 @@ mod authority_tests {
             let uri = upstream_uri(backend, "/block_refresh/block/10").unwrap();
             let mut headers = HeaderMap::new();
             headers.insert(HOST, HeaderValue::from_static("unitraklub.pl"));
-            let target = upstream_request_target(&uri, &mut headers, backend, false, false).unwrap();
+            let target = upstream_request_target(uri, &mut headers, backend, false, false).unwrap();
             assert_eq!(target.authority().unwrap().as_str(), expected_authority, "{protocol:?}");
             assert_eq!(target.path(), "/block_refresh/block/10");
             assert_eq!(headers.contains_key(HOST), !remove_host);
@@ -66,6 +66,27 @@ mod authority_tests {
         }
     }
 
+    #[test]
+    fn optimized_log_prefix_preserves_unicode_char_count_and_ascii_bytes() {
+        assert_eq!(log_prefix("/small.html", 256), "/small.html");
+        assert_eq!(log_prefix("abcdefghijklmnopqrstuvwxyz", 5), "abcde");
+        assert_eq!(log_prefix("za\u{17c}\u{f3}\u{142}\u{107}\u{1f525}", 4),
+                   "za\u{17c}\u{f3}");
+        assert_eq!(log_prefix("", 0), "");
+    }
+
+    #[test]
+    fn moved_uri_keeps_h1_origin_and_query() {
+        let config = crate::config::demo_config("127.0.0.1:9000".parse().unwrap());
+        let snapshot = crate::config::compile(config, 1, &validation_policy()).unwrap();
+        let backend = &snapshot.pools[0].members[0].backend;
+        let uri = upstream_uri(backend, "/keep?key=hello%2Fworld").unwrap();
+        let mut headers = HeaderMap::new();
+        let target = upstream_request_target(uri, &mut headers, backend, false, false).unwrap();
+        assert_eq!(target.path_and_query().unwrap().as_str(), "/keep?key=hello%2Fworld");
+        assert_eq!(target.authority().unwrap(), backend.request_origin_uri.authority().unwrap());
+    }
+
     #[test]
     fn invalid_outgoing_host_is_rejected_instead_of_used_for_dialing() {
         let mut config = crate::config::demo_config("127.0.0.1:9000".parse().unwrap());
@@ -76,7 +97,7 @@ mod authority_tests {
         let uri = upstream_uri(backend, "/").unwrap();
         let mut headers = HeaderMap::new();
         headers.insert(HOST, HeaderValue::from_static("evil@example.invalid"));
-        assert!(upstream_request_target(&uri, &mut headers, backend, false, false).is_none());
+        assert!(upstream_request_target(uri, &mut headers, backend, false, false).is_none());
     }
 
     #[test]

src/proxy/types.rs

@@ -180,7 +180,7 @@ fn legacy_websocket_origin(
 /// mode must retain the Host field: Hyper's legacy H1-compatible request may
 /// be sent over an H2-negotiated connection, and H1 fallback needs Host.
 fn upstream_request_target(
-    uri: &Uri,
+    uri: Uri,
     headers: &mut HeaderMap,
     backend: &crate::config::CompiledBackend,
     websocket_h1: bool,
@@ -193,7 +193,7 @@ fn upstream_request_target(
             crate::config::UpstreamProtocol::Http2 | crate::config::UpstreamProtocol::H2 => true,
         };
     if !h2_capable {
-        return Some(uri.clone());
+        return Some(uri);
     }
     let selected = headers.get(HOST).unwrap_or(&backend.host_header);
     let authority: http::uri::Authority = selected.to_str().ok()?.parse().ok()?;
@@ -201,7 +201,7 @@ fn upstream_request_target(
     if authority.host().is_empty() || authority.as_str().contains('@') {
         return None;
     }
-    let mut parts = uri.clone().into_parts();
+    let mut parts = uri.into_parts();
     parts.authority = Some(authority);
     let target = Uri::from_parts(parts).ok()?;
     let strict_h2_request = websocket_h2
@@ -259,7 +259,7 @@ fn retry_request(
             },
         );
     }
-    let target = upstream_request_target(&uri, request.headers_mut(), backend, false, false)?;
+    let target = upstream_request_target(uri, request.headers_mut(), backend, false, false)?;
     *request.uri_mut() = target;
     Some(request)
 }

src/proxy/upstream_helpers.rs

@@ -1,5 +1,20 @@
 include!("request_id.rs");
 
+// Retain the exchange through the response, but do not allocate an additional
+// body adapter for a request whose body has no frames or trailers.
+fn wrap_observed_request(body: Body, exchange: &Arc<Exchange>, idle: Duration) -> Body {
+    if body.is_end_stream() { body }
+    else { Body::new(MeteredBody::new(body, exchange.clone(), Direction::Request, idle)) }
+}
+
+fn log_prefix(input: &str, max_chars: usize) -> String {
+    if input.len() <= max_chars || input.is_ascii() {
+        input[..input.len().min(max_chars)].to_owned()
+    } else {
+        input.chars().take(max_chars).collect()
+    }
+}
+
 fn should_replace_error_page(response: &Response, minimal_welcome: bool) -> bool {
     !minimal_welcome
         && response
@@ -117,14 +132,12 @@ async fn observe(State(state): State<Arc<App>>, mut request: Request, next: Next
     if let Some(frontend) = &frontend_context {
         frontend.metrics.active.fetch_add(1, Ordering::Relaxed);
     }
-    let host = request
-        .headers()
-        .get(HOST)
-        .and_then(|h| h.to_str().ok())
-        .unwrap_or_else(|| request.uri().authority().map(|v| v.as_str()).unwrap_or(""))
-        .chars()
-        .take(253)
-        .collect();
+    let host = log_prefix(
+        request.headers().get(HOST)
+            .and_then(|h| h.to_str().ok())
+            .unwrap_or_else(|| request.uri().authority().map(|v| v.as_str()).unwrap_or("")),
+        253,
+    );
     let initial_client_ip = request
         .extensions()
         .get::<PeerAddr>()
@@ -164,7 +177,7 @@ async fn observe(State(state): State<Arc<App>>, mut request: Request, next: Next
         request_id: request_id.clone(),
         method: request.method().to_string(),
         host,
-        path: request.uri().path().chars().take(256).collect(),
+        path: log_prefix(request.uri().path(), 256),
         http_version,
         transport,
         websocket_bootstrap,
@@ -200,14 +213,7 @@ async fn observe(State(state): State<Arc<App>>, mut request: Request, next: Next
         .as_ref()
         .and_then(|frontend| frontend.proxy_timeouts.send_timeout())
         .unwrap_or(state.body_idle_timeout);
-    let request = request.map(|body| {
-        Body::new(MeteredBody::new(
-            body,
-            exchange.clone(),
-            Direction::Request,
-            request_send_timeout,
-        ))
-    });
+    let request = request.map(|body| wrap_observed_request(body, &exchange, request_send_timeout));
     let mut response = next.run(request).await;
     let final_route_context = response
         .extensions()