grpc/client: HTTP CONNECT proxy support in transport

grpc/Cargo.toml

@@ -55,6 +55,7 @@ bytes = "1.10.1"
 futures = { version = "0.3", default-features = false, optional = true }
 hickory-resolver = { version = "0.26.1", optional = true }
 http = "1.1.0"
+httparse = "1.9"
 http-body = "1.0.1"
 hyper = { version = "1.6.0", features = ["client", "http2"] }
 hyper-util = { version = "0.1", features = ["client-proxy"] }

grpc/src/client/channel.rs

@@ -67,6 +67,7 @@ use crate::client::name_resolution::ResolverUpdate;
 use crate::client::name_resolution::Target;
 use crate::client::name_resolution::dns;
 use crate::client::name_resolution::global_registry;
+use crate::client::name_resolution::proxy_resolver;
 use crate::client::name_resolution::{self};
 use crate::client::service_config::LbPolicyType;
 use crate::client::service_config::ServiceConfig;
@@ -248,6 +249,7 @@ impl PersistentChannel {
         // TODO(nathanielford): Return errors here instead of panicking.
         let target = Target::from_str(&target.into()).unwrap();
         let resolver_builder = global_registry().get(target.scheme()).unwrap();
+        let resolver_builder = Arc::new(proxy_resolver::Builder::new(resolver_builder));
         let authority = options
             .channel_authority
             .clone()

grpc/src/client/subchannel.rs

@@ -52,8 +52,10 @@ use crate::client::load_balancing::subchannel::private::Sealed;
 use crate::client::name_resolution::Address;
 use crate::client::stream_util::FailingRecvStream;
 use crate::client::transport::DynTransport;
+use crate::client::transport::ProxyOptions;
 use crate::client::transport::SecurityOpts;
 use crate::client::transport::TransportOptions;
+use crate::client::transport::http_connect::HttpConnectHandshaker;
 use crate::core::RequestHeaders;
 use crate::credentials::call::CallDetails;
 use crate::credentials::call::ClientConnectionSecurityInfo as CallClientConnectionSecurityInfo;
@@ -306,11 +308,18 @@ impl InternalSubchannel {
         transport: Arc<dyn DynTransport>,
         backoff: Arc<dyn Backoff>,
         runtime: GrpcRuntime,
-        security_opts: SecurityOpts,
+        mut security_opts: SecurityOpts,
         work_queue: WorkQueueTx,
     ) -> Arc<dyn Subchannel> {
         let on_drop = Arc::new(Notify::new());
         let address_string = address.address.to_string();
+        let transport_options = TransportOptions::default();
+        if let Some(proxy_opts) = ProxyOptions::from_addr(&address) {
+            security_opts.credentials = Arc::new(HttpConnectHandshaker::new(
+                security_opts.credentials,
+                proxy_opts,
+            ));
+        }
         let this = Arc::new_cyclic(|weak_self| Self {
             address,
             on_drop: on_drop.clone(),

grpc/src/client/transport/http_connect/mod.rs

@@ -0,0 +1,188 @@
+/*
+ *
+ * Copyright 2026 gRPC authors.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to
+ * deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+ * sell copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ *
+ */
+
+use std::sync::Arc;
+
+use bytes::Bytes;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tonic::async_trait;
+
+use crate::client::transport::ProxyOptions;
+use crate::client::transport::http_connect::rewind::Rewind;
+use crate::credentials::ChannelCredentials;
+use crate::credentials::ProtocolInfo;
+use crate::credentials::call::CallCredentials;
+use crate::credentials::client::ClientHandshakeInfo;
+use crate::credentials::client::HandshakeOutput;
+use crate::credentials::common::Authority;
+use crate::private;
+use crate::rt::AsyncIoAdapter;
+use crate::rt::BoxEndpoint;
+use crate::rt::GrpcEndpoint;
+use crate::rt::GrpcRuntime;
+use crate::rt::TokioIoStream;
+
+mod rewind;
+
+/// Performs the HTTP CONNECT handshake on the given endpoint.
+///
+/// This function sends an HTTP CONNECT request to the proxy specified in `opts`,
+/// reads the response, and returns a new endpoint that yields any buffered data
+/// read during the handshake before delegating to the original endpoint.
+async fn do_connect_handshake<I: GrpcEndpoint>(
+    input: I,
+    opts: &ProxyOptions,
+) -> Result<ProxyStream<I>, String> {
+    let mut io = AsyncIoAdapter::new(input);
+
+    let mut req = format!(
+        "CONNECT {} HTTP/1.1\r\nHost: {}\r\n",
+        opts.target_authority(),
+        opts.target_authority()
+    )
+    .into_bytes();
+    if let Some(creds) = opts.proxy_authorization_header() {
+        req.extend_from_slice(b"Proxy-Authorization: ");
+        req.extend_from_slice(creds.as_bytes());
+        req.extend_from_slice(b"\r\n");
+    }
+    req.extend_from_slice(b"\r\n"); // headers end
+
+    io.write_all(&req).await.map_err(|e| e.to_string())?;
+    io.flush().await.map_err(|e| e.to_string())?;
+
+    const READ_BUF_SIZE: usize = 8192;
+    let mut buf = vec![0u8; READ_BUF_SIZE];
+    let mut read = 0;
+
+    // Read the response.
+    loop {
+        let n = io.read(&mut buf[read..]).await.map_err(|e| e.to_string())?;
+        if n == 0 {
+            return Err("Connection closed by proxy".to_string());
+        }
+        read += n;
+
+        // Allocate space on the stack to read up to 16 headers from the proxy.
+        let mut headers = [httparse::EMPTY_HEADER; 16];
+        let mut res = httparse::Response::new(&mut headers);
+        match res.parse(&buf[..read]) {
+            Ok(httparse::Status::Complete(len)) => {
+                if res.code != Some(200) {
+                    return Err(format!("Proxy returned status {}", res.code.unwrap_or(0)));
+                }
+                // Success!
+                let remaining = read - len;
+                let buffered_data = if remaining > 0 {
+                    Some(Bytes::copy_from_slice(&buf[len..read]))
+                } else {
+                    None
+                };
+
+                let local_addr = io
+                    .get_ref()
+                    .get_local_address()
+                    .to_string()
+                    .into_boxed_str();
+                let peer_addr = io.get_ref().get_peer_address().to_string().into_boxed_str();
+                let network_type = io.get_ref().get_network_type();
+                // Check for buffered data. In most cases, the buffer should be
+                // empty as the server waits for the client to send the first
+                // message, e.g. in TLS.
+                let endpoint = if let Some(data) = buffered_data {
+                    Rewind::new_buffered(io, data)
+                } else {
+                    Rewind::new_unbuffered(io)
+                };
+                return Ok(TokioIoStream::new(
+                    endpoint,
+                    local_addr,
+                    peer_addr,
+                    network_type,
+                ));
+            }
+            Ok(httparse::Status::Partial) => {
+                if read >= READ_BUF_SIZE {
+                    return Err("Response too large".to_string());
+                }
+            }
+            Err(e) => {
+                return Err(format!("Failed to parse HTTP response: {}", e));
+            }
+        }
+    }
+}
+
+/// A credential wrapper that performs an HTTP CONNECT handshake before
+/// delegating to an inner security credential (like TLS).
+pub(crate) struct HttpConnectHandshaker {
+    inner: Arc<dyn ChannelCredentials>,
+    options: ProxyOptions,
+}
+
+impl HttpConnectHandshaker {
+    /// Constructs a new `ProxyChannelCredentials` wrapping the inner credentials.
+    pub(crate) fn new(inner: Arc<dyn ChannelCredentials>, options: &ProxyOptions) -> Self {
+        Self {
+            inner,
+            options: options.clone(),
+        }
+    }
+}
+
+/// The I/O stream wrapper returned after the HTTP CONNECT handshake succeeds.
+type ProxyStream<I> = TokioIoStream<Rewind<AsyncIoAdapter<I>>>;
+
+#[async_trait]
+impl ChannelCredentials for HttpConnectHandshaker {
+    fn info(&self) -> &ProtocolInfo {
+        self.inner.info()
+    }
+
+    fn get_call_credentials(&self, token: private::Internal) -> Option<&Arc<dyn CallCredentials>> {
+        self.inner.get_call_credentials(token)
+    }
+
+    async fn connect(
+        &self,
+        authority: &Authority,
+        source: BoxEndpoint,
+        info: &ClientHandshakeInfo,
+        runtime: &GrpcRuntime,
+        token: private::Internal,
+    ) -> Result<HandshakeOutput, String> {
+        // Perform the HTTP CONNECT handshake.
+        let proxied_stream = do_connect_handshake(source, &self.options).await?;
+
+        // Delegate the actual security handshake (e.g., TLS) to the wrapped
+        // credentials.
+        self.inner
+            .connect(authority, Box::new(proxied_stream), info, runtime, token)
+            .await
+    }
+}
+
+#[cfg(test)]
+mod test;

grpc/src/client/transport/http_connect/rewind.rs

@@ -0,0 +1,114 @@
+// Copyright (c) 2023-2025 Sean McArthur
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+//
+// Note: This file contains modifications by the gRPC authors; see revision
+// history for details.
+
+use std::cmp;
+use std::io;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+use bytes::Buf;
+use bytes::Bytes;
+use tokio::io::AsyncRead;
+use tokio::io::AsyncWrite;
+use tokio::io::ReadBuf;
+
+/// Combine a buffer with an IO, rewinding reads to use the buffer.
+#[derive(Debug)]
+pub(crate) struct Rewind<T> {
+    pub(crate) pre: Option<Bytes>,
+    pub(crate) inner: T,
+}
+
+impl<T> Rewind<T> {
+    pub(crate) fn new_buffered(io: T, buf: Bytes) -> Self {
+        Rewind {
+            pre: Some(buf),
+            inner: io,
+        }
+    }
+    pub(crate) fn new_unbuffered(io: T) -> Self {
+        Rewind {
+            pre: None,
+            inner: io,
+        }
+    }
+}
+
+impl<T> AsyncRead for Rewind<T>
+where
+    T: AsyncRead + Unpin,
+{
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        if let Some(mut prefix) = self.pre.take()
+            && !prefix.is_empty()
+        {
+            let copy_len = cmp::min(prefix.len(), buf.remaining());
+            buf.put_slice(&prefix[..copy_len]);
+            prefix.advance(copy_len);
+            if !prefix.is_empty() {
+                self.pre = Some(prefix);
+            }
+
+            return Poll::Ready(Ok(()));
+        }
+        Pin::new(&mut self.inner).poll_read(cx, buf)
+    }
+}
+
+impl<T> AsyncWrite for Rewind<T>
+where
+    T: AsyncWrite + Unpin,
+{
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.inner).poll_write(cx, buf)
+    }
+
+    fn poll_write_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[io::IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.inner).poll_write_vectored(cx, bufs)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.inner).poll_shutdown(cx)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        self.inner.is_write_vectored()
+    }
+}