fix(mesh): strip credentials before mesh peer forwarding (#1189)

crates/mesh-llm-host-runtime/src/network/openai/forwarded_request.rs

@@ -0,0 +1,394 @@
+//! Forwarded request construction for the proxy path.
+//!
+//! This module owns rebuilding an HTTP/1.x request before forwarding it to a
+//! peer or local inference target: parsing the Connection header to discover
+//! nominated hop-by-hop headers, rejecting framing-nominated Connection options
+//! per RFC 7230 §6.1, stripping omitted headers, and (for peer forwards) the
+//! caller credential headers.
+//!
+//! Behavior here is byte-faithful: header field values are preserved as raw
+//! octets, never coerced through UTF-8.
+
+use anyhow::{Context, Result, bail};
+
+use super::request_parse::MAX_HEADERS;
+use super::response::is_valid_header_name;
+
+/// Parse every `Connection` header on the request and return its nominated
+/// hop-by-hop header names.
+///
+/// Tokens are validated against HTTP/1.x token syntax (RFC 7230 §3.2.6
+/// `tchar`): every character must be in the complete tchar set, so
+/// separators, control characters, whitespace, and non-ASCII bytes all reject
+/// the request. Any malformed value rejects the entire request. Tokens are
+/// returned verbatim; callers compare case-insensitively via
+/// `eq_ignore_ascii_case` so neither this helper nor its caller pays for a
+/// per-token `to_lowercase()` allocation.
+fn collect_connection_nominated_tokens<'a>(
+    headers: &'a [httparse::Header<'a>],
+) -> Result<Vec<&'a str>> {
+    let mut nominated = Vec::new();
+    for h in headers {
+        if !h.name.eq_ignore_ascii_case("connection") {
+            continue;
+        }
+
+        // Connection values must be valid HTTP/1.x token syntax — ASCII only.
+        // Reject the entire request when invalid bytes or malformed tokens appear.
+        let value = std::str::from_utf8(h.value)
+            .context("Connection header contains non-UTF-8 bytes; reject as malformed")?;
+
+        for tok in value.split(',') {
+            // RFC 7230 §3.2.3: OWS is SP and HTAB only. Trimming just those
+            // means non-ASCII whitespace (e.g. NBSP) is never silently
+            // swallowed at a token edge — it fails the tchar check below.
+            let t = tok.trim_matches(|c| c == ' ' || c == '\t');
+            if t.is_empty() {
+                continue;
+            }
+            // Every character must be in the complete HTTP tchar set
+            // (RFC 7230 §3.2.6). Reuse the canonical tchar predicate so this
+            // parser and the response writers share one definition; it
+            // rejects separators (';', ':', ...), control characters, space,
+            // and any non-ASCII byte.
+            if !is_valid_header_name(t) {
+                bail!(
+                    "Connection header contains malformed token '{}'; reject as malformed",
+                    value
+                );
+            }
+            nominated.push(t);
+        }
+    }
+    Ok(nominated)
+}
+
+/// Reject framing-related hop-by-hop fields nominated via Connection.
+///
+/// Per RFC 7230 §6.1, `content-length` and `transfer-encoding` MUST NOT be
+/// used as Connection options because they are framing-related hop-by-hop
+/// fields with special semantics. Reject malformed requests before removing
+/// headers or forwarding the wire body.
+fn reject_framing_nominated(nominated: &[&str]) -> Result<()> {
+    if nominated
+        .iter()
+        .any(|n| n.eq_ignore_ascii_case("content-length"))
+    {
+        bail!(
+            "Connection header nominates 'Content-Length'; reject as protocol violation (RFC 7230 §6.1)"
+        );
+    }
+
+    if nominated
+        .iter()
+        .any(|n| n.eq_ignore_ascii_case("transfer-encoding"))
+    {
+        bail!(
+            "Connection header nominates 'Transfer-Encoding'; reject as protocol violation (RFC 7230 §6.1)"
+        );
+    }
+
+    Ok(())
+}
+
+pub(super) fn finalize_forwarded_request(
+    raw: &[u8],
+    strip_expect: bool,
+    rewritten_path: Option<&str>,
+    rewritten_body: Option<&[u8]>,
+    omitted_headers: &[&str],
+) -> Result<Vec<u8>> {
+    // Re-parse with httparse so we iterate over validated header structs.
+    let mut headers_buf = [httparse::EMPTY_HEADER; MAX_HEADERS];
+    let mut req = httparse::Request::new(&mut headers_buf);
+    let header_end = match req.parse(raw).context("re-parse headers for forwarding")? {
+        httparse::Status::Complete(header_end) => header_end,
+        httparse::Status::Partial => bail!("incomplete HTTP headers for forwarding"),
+    };
+
+    let method = req.method.unwrap_or("GET");
+    let path = rewritten_path.unwrap_or_else(|| req.path.unwrap_or("/"));
+    let version = req.version.unwrap_or(1);
+
+    let mut rebuilt = Vec::new();
+    rebuilt.extend_from_slice(format!("{method} {path} HTTP/1.{version}\r\n").as_bytes());
+
+    // RFC 7230 §6.1: parse Connection header to find nominated hop-by-hop
+    // headers, and reject framing-related nominations before forwarding.
+    let connection_nominated = collect_connection_nominated_tokens(req.headers)?;
+    reject_framing_nominated(&connection_nominated)?;
+
+    for header in req.headers.iter() {
+        let name = header.name;
+        // Compare case-insensitively against each ASCII header name without per-header allocation.
+        let is_connection_nominated = connection_nominated
+            .iter()
+            .any(|n| n.eq_ignore_ascii_case(name));
+        if name.eq_ignore_ascii_case("connection")
+            || is_connection_nominated
+            || omitted_headers
+                .iter()
+                .any(|omitted| name.eq_ignore_ascii_case(omitted))
+        {
+            continue;
+        }
+        if strip_expect && name.eq_ignore_ascii_case("expect") {
+            continue;
+        }
+        if rewritten_body.is_some()
+            && (name.eq_ignore_ascii_case("content-length")
+                || name.eq_ignore_ascii_case("transfer-encoding"))
+        {
+            continue;
+        }
+        // Preserve header values as raw bytes — HTTP/1.x field-values are arbitrary
+        // octets, not UTF-8 strings. Write name (ASCII) + ": " separator + raw
+        // value, then terminate with CRLF.
+        rebuilt.extend_from_slice(name.as_ref());
+        rebuilt.push(b':');
+        rebuilt.push(b' ');
+        rebuilt.extend_from_slice(header.value);
+        rebuilt.extend_from_slice(b"\r\n");
+    }
+    if let Some(body) = rewritten_body {
+        rebuilt.extend_from_slice(format!("Content-Length: {}\r\n", body.len()).as_bytes());
+    }
+
+    // The proxy buffers exactly one request for routing, so force a single-request connection contract upstream instead of reusing the client connection blindly.
+    rebuilt.extend_from_slice(b"Connection: close\r\n\r\n");
+
+    let mut forwarded = rebuilt;
+    forwarded.extend_from_slice(rewritten_body.unwrap_or(&raw[header_end..]));
+    Ok(forwarded)
+}
+
+/// Rebuild a request for a remote peer without forwarding ingress credentials.
+pub(super) fn prepare_peer_forwarded_request(raw: &[u8]) -> Result<Vec<u8>> {
+    const CALLER_CREDENTIAL_HEADERS: &[&str] = &[
+        "authorization",
+        "proxy-authorization",
+        "x-api-key",
+        "api-key",
+    ];
+    finalize_forwarded_request(raw, false, None, None, CALLER_CREDENTIAL_HEADERS)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn peer_forwarding_strips_credentials_and_preserves_headers_and_body() {
+        let body = br#"{"model":"test","input":"Authorization: keep in body"}"#;
+        let mut raw = format!(
+            "POST /v1/responses HTTP/1.1\r\nHost: localhost\r\naUtHoRiZaTiOn: Bearer caller-secret\r\nX-Trace-Id: trace-123\r\nPROXY-AUTHORIZATION: Basic proxy-secret\r\nX-API-Key: anthropic-secret\r\napi-key: azure-secret\r\nContent-Length: {}\r\n\r\n",
+            body.len()
+        )
+        .into_bytes();
+        raw.extend_from_slice(body);
+
+        let forwarded = prepare_peer_forwarded_request(&raw).unwrap();
+
+        let mut expected = format!(
+            "POST /v1/responses HTTP/1.1\r\nHost: localhost\r\nX-Trace-Id: trace-123\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+            body.len()
+        )
+        .into_bytes();
+        expected.extend_from_slice(body);
+        assert_eq!(forwarded, expected);
+    }
+
+    #[test]
+    fn peer_forwarding_rejects_incomplete_headers() {
+        let raw = b"GET /v1/models HTTP/1.1\r\nAuthorization: Bearer caller-secret\r\n";
+
+        assert!(prepare_peer_forwarded_request(raw).is_err());
+    }
+
+    #[test]
+    fn finalize_strips_connection_nominated_headers() {
+        let raw = b"GET /v1/chat/completions HTTP/1.1\r\nHost: localhost:9337\r\nConnection: TE, Upgrade\r\nTE: trailers\r\nUpgrade: h2c\r\nX-Keep: yes\r\nContent-Length: 0\r\n\r\n";
+        let result = finalize_forwarded_request(raw, false, None, None, &[]).unwrap();
+        let text = String::from_utf8_lossy(&result);
+
+        // Proxy adds exactly one Connection header of its own.
+        assert_eq!(text.matches("Connection:").count(), 1);
+
+        // TE and Upgrade nominated by Connection must be stripped (RFC 7230 §6.1 fix)
+        assert!(
+            !text.contains("TE:"),
+            "expected TE stripped via Connection nomination"
+        );
+        assert!(
+            !text.contains("Upgrade:"),
+            "expected Upgrade stripped via Connection nomination"
+        );
+
+        // Non-nominated headers preserved, proxy adds its own Connection: close
+        assert!(text.contains("X-Keep: yes"), "expected X-Keep preserved");
+    }
+
+    #[test]
+    fn finalize_connection_nominated_case_insensitive() {
+        let raw = b"GET / HTTP/1.1\r\nHost: x\r\nConnection: te, upgrade\r\nte: trailers\r\nUpgrade: h2c\r\nContent-Length: 0\r\n\r\n";
+        let result = finalize_forwarded_request(raw, false, None, None, &[]).unwrap();
+        let text = String::from_utf8_lossy(&result);
+
+        assert!(!text.contains("te:"), "expected te stripped");
+        assert!(
+            !text.to_lowercase().contains("upgrade"),
+            "expected Upgrade stripped case-insensitively"
+        );
+    }
+
+    /// Regression: RFC 7230 §6.1 forbids nominating Content-Length via Connection.
+    /// The request MUST be rejected before header removal or body forwarding.
+    #[test]
+    fn finalize_rejects_connection_nominating_content_length() {
+        let raw =
+            b"GET / HTTP/1.1\r\nHost: x\r\nConnection: Content-Length\r\nContent-Length: 0\r\n\r\n";
+        let result = finalize_forwarded_request(raw, false, None, None, &[]);
+        assert!(
+            result.is_err(),
+            "nominating Content-Length via Connection must be rejected, got: {:?}",
+            result
+        );
+        let msg = result.unwrap_err().to_string();
+        assert!(
+            msg.contains("Content-Length"),
+            "error should name Content-Length framing violation, got: {msg}"
+        );
+    }
+
+    /// Regression: RFC 7230 §6.1 forbids nominating Transfer-Encoding via Connection.
+    /// The request MUST be rejected before header removal or body forwarding.
+    #[test]
+    fn finalize_rejects_connection_nominating_transfer_encoding() {
+        let raw = b"GET / HTTP/1.1\r\nHost: x\r\nConnection: Transfer-Encoding\r\nTransfer-Encoding: chunked\r\nContent-Length: 0\r\n\r\n";
+        let result = finalize_forwarded_request(raw, false, None, None, &[]);
+        assert!(
+            result.is_err(),
+            "nominating Transfer-Encoding via Connection must be rejected, got: {:?}",
+            result
+        );
+        let msg = result.unwrap_err().to_string();
+        assert!(
+            msg.contains("Transfer-Encoding"),
+            "error should name Transfer-Encoding framing violation, got: {msg}"
+        );
+    }
+
+    /// Regression: header field values are arbitrary octets (RFC 7230 §3.2).
+    /// A non-UTF-8 value on an ordinary header must be preserved verbatim, not
+    /// coerced to an empty string.
+    #[test]
+    fn finalize_preserves_opaque_non_utf8_field_value() {
+        // X-Custom carries a raw ISO-8859-1 byte 0xE9 ('é') which is invalid UTF-8.
+        let raw: &[u8] =
+            b"GET / HTTP/1.1\r\nHost: x\r\nX-Custom: caf\xE9\r\nContent-Length: 0\r\n\r\n";
+        let result = finalize_forwarded_request(raw, false, None, None, &[]).unwrap();
+        // The raw 0xE9 byte must appear unchanged in the rebuilt request.
+        assert!(
+            result.windows(1).any(|w| w == [0xE9]),
+            "opaque non-UTF-8 field value must be preserved as raw bytes, rebuilt: {:?}",
+            String::from_utf8_lossy(&result)
+        );
+        assert!(
+            result
+                .windows(b"X-Custom: ".len())
+                .any(|w| w == b"X-Custom: "),
+            "header name + value must be present in the rebuilt request"
+        );
+    }
+
+    /// Regression: a malformed Connection value containing invalid bytes must be
+    /// rejected wholesale, not partially filtered. A high-bit byte is invalid in
+    /// HTTP/1.x token syntax and must fail the request.
+    #[test]
+    fn finalize_rejects_malformed_connection_value() {
+        // 0xFF inside the Connection token list — invalid ASCII token syntax.
+        let raw: &[u8] = b"GET / HTTP/1.1\r\nHost: x\r\nConnection: keep-alive, \xFF\r\nContent-Length: 0\r\n\r\n";
+        let result = finalize_forwarded_request(raw, false, None, None, &[]);
+        assert!(
+            result.is_err(),
+            "malformed Connection value with invalid bytes must be rejected, got: {:?}",
+            result
+        );
+    }
+
+    /// Regression: a Connection value that is valid UTF-8 but contains an ASCII
+    /// control char is still malformed token syntax and must be rejected.
+    #[test]
+    fn finalize_rejects_connection_control_char() {
+        let raw =
+            b"GET / HTTP/1.1\r\nHost: x\r\nConnection: keep-alive\x01\r\nContent-Length: 0\r\n\r\n";
+        let result = finalize_forwarded_request(raw, false, None, None, &[]);
+        assert!(
+            result.is_err(),
+            "Connection with control character must be rejected, got: {:?}",
+            result
+        );
+    }
+
+    /// Regression: a Connection value that is valid UTF-8 but contains a
+    /// non-ASCII character (high-bit byte that decodes cleanly, e.g. a UTF-8
+    /// multibyte sequence) is still invalid HTTP/1.x token syntax and must be
+    /// rejected. This exercises the non-ASCII token validation branch rather
+    /// than the invalid-UTF-8 path covered by
+    /// `finalize_rejects_malformed_connection_value`.
+    #[test]
+    fn finalize_rejects_connection_non_ascii_token() {
+        // 'é' encodes as 0xC3 0xA9 in UTF-8: valid UTF-8, non-ASCII token syntax.
+        let raw: &[u8] =
+            b"GET / HTTP/1.1\r\nHost: x\r\nConnection: keep-alive, caf\xC3\xA9\r\nContent-Length: 0\r\n\r\n";
+        let result = finalize_forwarded_request(raw, false, None, None, &[]);
+        assert!(
+            result.is_err(),
+            "Connection with non-ASCII but valid-UTF-8 token must be rejected, got: {:?}",
+            result
+        );
+        let msg = result.unwrap_err().to_string();
+        assert!(
+            msg.contains("malformed token"),
+            "error should report the malformed token, got: {msg}"
+        );
+    }
+
+    /// Regression: Connection options must be tokens (RFC 7230 §3.2.6).
+    /// Separator characters such as ';' are not tchars and must reject the
+    /// request instead of being accepted as part of a token.
+    #[test]
+    fn finalize_rejects_connection_separator_char() {
+        let raw =
+            b"GET / HTTP/1.1\r\nHost: x\r\nConnection: keep-alive;foo\r\nContent-Length: 0\r\n\r\n";
+        let result = finalize_forwarded_request(raw, false, None, None, &[]);
+        assert!(
+            result.is_err(),
+            "Connection option containing a separator must be rejected, got: {:?}",
+            result
+        );
+        let msg = result.unwrap_err().to_string();
+        assert!(
+            msg.contains("malformed token"),
+            "error should report the malformed token, got: {msg}"
+        );
+    }
+
+    /// Regression: whitespace around Connection options is OWS, i.e. only SP
+    /// and HTAB (RFC 7230 §3.2.3). Non-ASCII whitespace such as NBSP must not
+    /// be trimmed at a token edge; it must fail the tchar check instead of
+    /// being silently accepted.
+    #[test]
+    fn finalize_rejects_connection_non_ascii_whitespace() {
+        // NBSP (U+00A0) between the comma and the next option: valid UTF-8,
+        // but neither SP/HTAB nor a tchar.
+        let raw: &[u8] =
+            b"GET / HTTP/1.1\r\nHost: x\r\nConnection: keep-alive, \xC2\xA0close\r\nContent-Length: 0\r\n\r\n";
+        let result = finalize_forwarded_request(raw, false, None, None, &[]);
+        assert!(
+            result.is_err(),
+            "Connection with non-ASCII whitespace must be rejected, got: {:?}",
+            result
+        );
+    }
+}

crates/mesh-llm-host-runtime/src/network/openai/mod.rs

@@ -1,4 +1,5 @@
 pub(crate) mod auto_route;
+mod forwarded_request;
 pub(crate) mod ingress;
 pub(crate) mod moa_gateway;
 mod request_normalize;

crates/mesh-llm-host-runtime/src/network/openai/request_parse.rs

@@ -5,6 +5,7 @@ use serde::Deserialize;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
 
+use super::forwarded_request::finalize_forwarded_request;
 use super::request_normalize::{
     ResponseAdapter, normalize_openai_compat_request, resolve_request_object_references,
 };
@@ -191,11 +192,11 @@ pub(super) async fn read_http_request_with_limits(
             .or(value.n_predict)
     });
     let raw = finalize_forwarded_request(
-        raw,
-        header_end,
+        &raw,
         parsed.expects_continue,
         Some(&rewrite.request_path),
         rewrite.rewritten_body.as_deref(),
+        &[],
     )?;
     let body_len_bytes = body.len();
     let body_bytes = if body.is_empty() { None } else { Some(body) };
@@ -354,55 +355,6 @@ fn body_limits_for_path(path: &str, default: HttpReadLimits) -> HttpReadLimits {
     }
 }
 
-fn finalize_forwarded_request(
-    mut raw: Vec<u8>,
-    header_end: usize,
-    strip_expect: bool,
-    rewritten_path: Option<&str>,
-    rewritten_body: Option<&[u8]>,
-) -> Result<Vec<u8>> {
-    let original_body = raw.split_off(header_end);
-    // Re-parse with httparse so we iterate over validated header structs.
-    let mut headers_buf = [httparse::EMPTY_HEADER; MAX_HEADERS];
-    let mut req = httparse::Request::new(&mut headers_buf);
-    let _ = req.parse(&raw).context("re-parse headers for forwarding")?;
-
-    let method = req.method.unwrap_or("GET");
-    let path = rewritten_path.unwrap_or_else(|| req.path.unwrap_or("/"));
-    let version = req.version.unwrap_or(1);
-
-    let mut rebuilt = format!("{method} {path} HTTP/1.{version}\r\n");
-
-    for header in req.headers.iter() {
-        let name = header.name;
-        if name.eq_ignore_ascii_case("connection") {
-            continue;
-        }
-        if strip_expect && name.eq_ignore_ascii_case("expect") {
-            continue;
-        }
-        if rewritten_body.is_some()
-            && (name.eq_ignore_ascii_case("content-length")
-                || name.eq_ignore_ascii_case("transfer-encoding"))
-        {
-            continue;
-        }
-        let value = std::str::from_utf8(header.value).unwrap_or("");
-        rebuilt.push_str(&format!("{name}: {value}\r\n"));
-    }
-    if let Some(body) = rewritten_body {
-        rebuilt.push_str(&format!("Content-Length: {}\r\n", body.len()));
-    }
-
-    // The proxy buffers exactly one request for routing, so force a single-request
-    // connection contract upstream instead of reusing the client connection blindly.
-    rebuilt.push_str("Connection: close\r\n\r\n");
-
-    let mut forwarded = rebuilt.into_bytes();
-    forwarded.extend_from_slice(rewritten_body.unwrap_or(&original_body));
-    Ok(forwarded)
-}
-
 /// Read from the stream until httparse can fully parse the request headers.
 /// Returns parsed metadata; `buf` contains all bytes read so far (headers +
 /// any trailing body bytes that arrived in the same read).

crates/mesh-llm-host-runtime/src/network/openai/response.rs

@@ -20,6 +20,6 @@ pub(crate) use models::send_models_list_with_descriptors;
 pub use pipeline::{PipelineProxyResult, pipeline_proxy_local};
 pub(super) use routing::{route_local_attempt, route_remote_attempt};
 pub(crate) use send::{
-    append_safe_header, send_400, send_503, send_error, send_json_ok, send_json_ok_with_headers,
-    send_json_with_status_and_headers,
+    append_safe_header, is_valid_header_name, send_400, send_503, send_error, send_json_ok,
+    send_json_ok_with_headers, send_json_with_status_and_headers,
 };

crates/mesh-llm-host-runtime/src/network/openai/response/routing.rs

@@ -2,6 +2,7 @@ use super::common::{ResponseRetryPolicy, RouteAttemptResult, retryable_route_res
 use super::dispatch::relay_attempted_response;
 use super::probe::{probe_http_response, probe_http_response_local};
 use crate::mesh;
+use crate::network::openai::forwarded_request::prepare_peer_forwarded_request;
 use crate::network::openai::request_normalize::ResponseAdapter;
 use tokio::io::AsyncWriteExt;
 use tokio::net::TcpStream;
@@ -95,32 +96,60 @@ pub(in crate::network::openai) async fn route_remote_attempt(
     retry_policy: ResponseRetryPolicy,
     response_adapter: ResponseAdapter,
 ) -> RouteAttemptResult {
-    match node.open_http_tunnel(host_id).await {
-        Ok((mut quic_send, mut quic_recv)) => {
-            if let Err(err) = quic_send.write_all(prefetched).await {
-                tracing::warn!(
-                    "API proxy: failed to forward buffered request to host {}: {err}",
-                    host_id.fmt_short()
-                );
-                return RouteAttemptResult::RetryableUnavailable;
-            }
-            route_remote_attempt_after_forward(
-                tcp_stream,
-                &mut quic_recv,
-                host_id,
-                retry_policy,
-                response_adapter,
-            )
-            .await
-        }
+    let (mut quic_send, mut quic_recv) = match node.open_http_tunnel(host_id).await {
+        Ok(tunnel) => tunnel,
         Err(err) => {
             tracing::warn!(
                 "API proxy: can't tunnel to host {}: {err}",
                 host_id.fmt_short()
             );
-            retryable_route_result_from_error(&err)
+            return retryable_route_result_from_error(&err);
         }
+    };
+
+    if forward_peer_request(&mut quic_send, host_id, prefetched)
+        .await
+        .is_err()
+    {
+        return RouteAttemptResult::RetryableUnavailable;
     }
+
+    route_remote_attempt_after_forward(
+        tcp_stream,
+        &mut quic_recv,
+        host_id,
+        retry_policy,
+        response_adapter,
+    )
+    .await
+}
+
+async fn forward_peer_request(
+    quic_send: &mut iroh::endpoint::SendStream,
+    host_id: iroh::EndpointId,
+    prefetched: &[u8],
+) -> Result<(), ()> {
+    // Caller credentials are meaningful at ingress, not on a remote peer.
+    let peer_request = match prepare_peer_forwarded_request(prefetched) {
+        Ok(request) => request,
+        Err(err) => {
+            tracing::warn!(
+                "API proxy: refusing to forward malformed request to host {}: {err}",
+                host_id.fmt_short()
+            );
+            return Err(());
+        }
+    };
+
+    if let Err(err) = quic_send.write_all(&peer_request).await {
+        tracing::warn!(
+            "API proxy: failed to forward buffered request to host {}: {err}",
+            host_id.fmt_short()
+        );
+        return Err(());
+    }
+
+    Ok(())
 }
 
 async fn route_remote_attempt_after_forward(