client: Add support for HTTP/HTTPS proxy (#30812)

Closes #30732

I tested it on my machine, and the HTTP proxy is working properly now.

Release Notes:

- N/A

Cargo.lock

@@ -2792,6 +2792,7 @@ dependencies = [
  "anyhow",
  "async-recursion 0.3.2",
  "async-tungstenite",
+ "base64 0.22.1",
  "chrono",
  "clock",
  "cocoa 0.26.0",
@@ -2803,6 +2804,7 @@ dependencies = [
  "gpui_tokio",
  "http_client",
  "http_client_tls",
+ "httparse",
  "log",
  "parking_lot",
  "paths",
@@ -2823,6 +2825,7 @@ dependencies = [
  "time",
  "tiny_http",
  "tokio",
+ "tokio-native-tls",
  "tokio-socks",
  "url",
  "util",

crates/client/Cargo.toml

@@ -19,6 +19,7 @@ test-support = ["clock/test-support", "collections/test-support", "gpui/test-sup
 anyhow.workspace = true
 async-recursion = "0.3"
 async-tungstenite = { workspace = true, features = ["tokio", "tokio-rustls-manual-roots"] }
+base64.workspace = true
 chrono = { workspace = true, features = ["serde"] }
 clock.workspace = true
 collections.workspace = true
@@ -29,6 +30,7 @@ gpui.workspace = true
 gpui_tokio.workspace = true
 http_client.workspace = true
 http_client_tls.workspace = true
+httparse = "1.10"
 log.workspace = true
 paths.workspace = true
 parking_lot.workspace = true
@@ -47,6 +49,7 @@ text.workspace = true
 thiserror.workspace = true
 time.workspace = true
 tiny_http = "0.8"
+tokio-native-tls = "0.3"
 tokio-socks = { version = "0.5.2", default-features = false, features = ["futures-io"] }
 url.workspace = true
 util.workspace = true

crates/client/src/client.rs

@@ -1,7 +1,7 @@
 #[cfg(any(test, feature = "test-support"))]
 pub mod test;
 
-mod socks;
+mod proxy;
 pub mod telemetry;
 pub mod user;
 pub mod zed_urls;
@@ -24,13 +24,13 @@ use gpui::{App, AsyncApp, Entity, Global, Task, WeakEntity, actions};
 use http_client::{AsyncBody, HttpClient, HttpClientWithUrl};
 use parking_lot::RwLock;
 use postage::watch;
+use proxy::connect_proxy_stream;
 use rand::prelude::*;
 use release_channel::{AppVersion, ReleaseChannel};
 use rpc::proto::{AnyTypedEnvelope, EnvelopedMessage, PeerId, RequestMessage};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use settings::{Settings, SettingsSources};
-use socks::connect_socks_proxy_stream;
 use std::pin::Pin;
 use std::{
     any::TypeId,
@@ -1156,7 +1156,7 @@ impl Client {
                 let handle = cx.update(|cx| gpui_tokio::Tokio::handle(cx)).ok().unwrap();
                 let _guard = handle.enter();
                 match proxy {
-                    Some(proxy) => connect_socks_proxy_stream(&proxy, rpc_host).await?,
+                    Some(proxy) => connect_proxy_stream(&proxy, rpc_host).await?,
                     None => Box::new(TcpStream::connect(rpc_host).await?),
                 }
             };

crates/client/src/proxy.rs

@@ -0,0 +1,66 @@
+//! client proxy
+
+mod http_proxy;
+mod socks_proxy;
+
+use anyhow::{Context, Result, anyhow};
+use http_client::Url;
+use http_proxy::{HttpProxyType, connect_http_proxy_stream, parse_http_proxy};
+use socks_proxy::{SocksVersion, connect_socks_proxy_stream, parse_socks_proxy};
+
+pub(crate) async fn connect_proxy_stream(
+    proxy: &Url,
+    rpc_host: (&str, u16),
+) -> Result<Box<dyn AsyncReadWrite>> {
+    let Some(((proxy_domain, proxy_port), proxy_type)) = parse_proxy_type(proxy) else {
+        // If parsing the proxy URL fails, we must avoid falling back to an insecure connection.
+        // SOCKS proxies are often used in contexts where security and privacy are critical,
+        // so any fallback could expose users to significant risks.
+        return Err(anyhow!("Parsing proxy url failed"));
+    };
+
+    // Connect to proxy and wrap protocol later
+    let stream = tokio::net::TcpStream::connect((proxy_domain.as_str(), proxy_port))
+        .await
+        .context("Failed to connect to proxy")?;
+
+    let proxy_stream = match proxy_type {
+        ProxyType::SocksProxy(proxy) => connect_socks_proxy_stream(stream, proxy, rpc_host).await?,
+        ProxyType::HttpProxy(proxy) => {
+            connect_http_proxy_stream(stream, proxy, rpc_host, &proxy_domain).await?
+        }
+    };
+
+    Ok(proxy_stream)
+}
+
+enum ProxyType<'t> {
+    SocksProxy(SocksVersion<'t>),
+    HttpProxy(HttpProxyType<'t>),
+}
+
+fn parse_proxy_type<'t>(proxy: &'t Url) -> Option<((String, u16), ProxyType<'t>)> {
+    let scheme = proxy.scheme();
+    let host = proxy.host()?.to_string();
+    let port = proxy.port_or_known_default()?;
+    let proxy_type = match scheme {
+        scheme if scheme.starts_with("socks") => {
+            Some(ProxyType::SocksProxy(parse_socks_proxy(scheme, proxy)))
+        }
+        scheme if scheme.starts_with("http") => {
+            Some(ProxyType::HttpProxy(parse_http_proxy(scheme, proxy)))
+        }
+        _ => None,
+    }?;
+
+    Some(((host, port), proxy_type))
+}
+
+pub(crate) trait AsyncReadWrite:
+    tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static
+{
+}
+impl<T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static> AsyncReadWrite
+    for T
+{
+}

crates/client/src/proxy/http_proxy.rs

@@ -0,0 +1,171 @@
+use anyhow::{Context, Result};
+use base64::Engine;
+use httparse::{EMPTY_HEADER, Response};
+use tokio::{
+    io::{AsyncBufReadExt, AsyncWriteExt, BufStream},
+    net::TcpStream,
+};
+use tokio_native_tls::{TlsConnector, native_tls};
+use url::Url;
+
+use super::AsyncReadWrite;
+
+pub(super) enum HttpProxyType<'t> {
+    HTTP(Option<HttpProxyAuthorization<'t>>),
+    HTTPS(Option<HttpProxyAuthorization<'t>>),
+}
+
+pub(super) struct HttpProxyAuthorization<'t> {
+    username: &'t str,
+    password: &'t str,
+}
+
+pub(super) fn parse_http_proxy<'t>(scheme: &str, proxy: &'t Url) -> HttpProxyType<'t> {
+    let auth = proxy.password().map(|password| HttpProxyAuthorization {
+        username: proxy.username(),
+        password,
+    });
+    if scheme.starts_with("https") {
+        HttpProxyType::HTTPS(auth)
+    } else {
+        HttpProxyType::HTTP(auth)
+    }
+}
+
+pub(crate) async fn connect_http_proxy_stream(
+    stream: TcpStream,
+    http_proxy: HttpProxyType<'_>,
+    rpc_host: (&str, u16),
+    proxy_domain: &str,
+) -> Result<Box<dyn AsyncReadWrite>> {
+    match http_proxy {
+        HttpProxyType::HTTP(auth) => http_connect(stream, rpc_host, auth).await,
+        HttpProxyType::HTTPS(auth) => https_connect(stream, rpc_host, auth, proxy_domain).await,
+    }
+    .context("error connecting to http/https proxy")
+}
+
+async fn http_connect<T>(
+    stream: T,
+    target: (&str, u16),
+    auth: Option<HttpProxyAuthorization<'_>>,
+) -> Result<Box<dyn AsyncReadWrite>>
+where
+    T: AsyncReadWrite,
+{
+    let mut stream = BufStream::new(stream);
+    let request = make_request(target, auth);
+    stream.write_all(request.as_bytes()).await?;
+    stream.flush().await?;
+    check_response(&mut stream).await?;
+    Ok(Box::new(stream))
+}
+
+async fn https_connect<T>(
+    stream: T,
+    target: (&str, u16),
+    auth: Option<HttpProxyAuthorization<'_>>,
+    proxy_domain: &str,
+) -> Result<Box<dyn AsyncReadWrite>>
+where
+    T: AsyncReadWrite,
+{
+    let tls_connector = TlsConnector::from(native_tls::TlsConnector::new()?);
+    let stream = tls_connector.connect(proxy_domain, stream).await?;
+    http_connect(stream, target, auth).await
+}
+
+fn make_request(target: (&str, u16), auth: Option<HttpProxyAuthorization<'_>>) -> String {
+    let (host, port) = target;
+    let mut request = format!(
+        "CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\nProxy-Connection: Keep-Alive\r\n"
+    );
+    if let Some(HttpProxyAuthorization { username, password }) = auth {
+        let auth =
+            base64::prelude::BASE64_STANDARD.encode(format!("{username}:{password}").as_bytes());
+        let auth = format!("Proxy-Authorization: Basic {auth}\r\n");
+        request.push_str(&auth);
+    }
+    request.push_str("\r\n");
+    request
+}
+
+async fn check_response<T>(stream: &mut BufStream<T>) -> Result<()>
+where
+    T: AsyncReadWrite,
+{
+    let response = recv_response(stream).await?;
+    let mut dummy_headers = [EMPTY_HEADER; MAX_RESPONSE_HEADERS];
+    let mut parser = Response::new(&mut dummy_headers);
+    parser.parse(response.as_bytes())?;
+
+    match parser.code {
+        Some(code) => {
+            if code == 200 {
+                Ok(())
+            } else {
+                Err(anyhow::anyhow!(
+                    "Proxy connection failed with HTTP code: {code}"
+                ))
+            }
+        }
+        None => Err(anyhow::anyhow!(
+            "Proxy connection failed with no HTTP code: {}",
+            parser.reason.unwrap_or("Unknown reason")
+        )),
+    }
+}
+
+const MAX_RESPONSE_HEADER_LENGTH: usize = 4096;
+const MAX_RESPONSE_HEADERS: usize = 16;
+
+async fn recv_response<T>(stream: &mut BufStream<T>) -> Result<String>
+where
+    T: AsyncReadWrite,
+{
+    let mut response = String::new();
+    loop {
+        if stream.read_line(&mut response).await? == 0 {
+            return Err(anyhow::anyhow!("End of stream"));
+        }
+
+        if MAX_RESPONSE_HEADER_LENGTH < response.len() {
+            return Err(anyhow::anyhow!("Maximum response header length exceeded"));
+        }
+
+        if response.ends_with("\r\n\r\n") {
+            return Ok(response);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use url::Url;
+
+    use super::{HttpProxyAuthorization, HttpProxyType, parse_http_proxy};
+
+    #[test]
+    fn test_parse_http_proxy() {
+        let proxy = Url::parse("http://proxy.example.com:1080").unwrap();
+        let scheme = proxy.scheme();
+
+        let version = parse_http_proxy(scheme, &proxy);
+        assert!(matches!(version, HttpProxyType::HTTP(None)))
+    }
+
+    #[test]
+    fn test_parse_http_proxy_with_auth() {
+        let proxy = Url::parse("http://username:password@proxy.example.com:1080").unwrap();
+        let scheme = proxy.scheme();
+
+        let version = parse_http_proxy(scheme, &proxy);
+        assert!(matches!(
+            version,
+            HttpProxyType::HTTP(Some(HttpProxyAuthorization {
+                username: "username",
+                password: "password"
+            }))
+        ))
+    }
+}

crates/client/src/proxy/socks_proxy.rs

@@ -1,15 +1,19 @@
 //! socks proxy
-use anyhow::{Context, Result, anyhow};
+
-use http_client::Url;
+use anyhow::{Context, Result};
+use tokio::net::TcpStream;
 use tokio_socks::tcp::{Socks4Stream, Socks5Stream};
+use url::Url;
+
+use super::AsyncReadWrite;
 
 /// Identification to a Socks V4 Proxy
-struct Socks4Identification<'a> {
+pub(super) struct Socks4Identification<'a> {
     user_id: &'a str,
 }
 
 /// Authorization to a Socks V5 Proxy
-struct Socks5Authorization<'a> {
+pub(super) struct Socks5Authorization<'a> {
     username: &'a str,
     password: &'a str,
 }
@@ -18,45 +22,50 @@ struct Socks5Authorization<'a> {
 ///
 /// V4 allows idenfication using a user_id
 /// V5 allows authorization using a username and password
-enum SocksVersion<'a> {
+pub(super) enum SocksVersion<'a> {
     V4(Option<Socks4Identification<'a>>),
     V5(Option<Socks5Authorization<'a>>),
 }
 
-pub(crate) async fn connect_socks_proxy_stream(
+pub(super) fn parse_socks_proxy<'t>(scheme: &str, proxy: &'t Url) -> SocksVersion<'t> {
-    proxy: &Url,
+    if scheme.starts_with("socks4") {
+        let identification = match proxy.username() {
+            "" => None,
+            username => Some(Socks4Identification { user_id: username }),
+        };
+        SocksVersion::V4(identification)
+    } else {
+        let authorization = proxy.password().map(|password| Socks5Authorization {
+            username: proxy.username(),
+            password,
+        });
+        SocksVersion::V5(authorization)
+    }
+}
+
+pub(super) async fn connect_socks_proxy_stream(
+    stream: TcpStream,
+    socks_version: SocksVersion<'_>,
     rpc_host: (&str, u16),
 ) -> Result<Box<dyn AsyncReadWrite>> {
-    let Some((socks_proxy, version)) = parse_socks_proxy(proxy) else {
+    match socks_version {
-        // If parsing the proxy URL fails, we must avoid falling back to an insecure connection.
-        // SOCKS proxies are often used in contexts where security and privacy are critical,
-        // so any fallback could expose users to significant risks.
-        return Err(anyhow!("Parsing proxy url failed"));
-    };
-
-    // Connect to proxy and wrap protocol later
-    let stream = tokio::net::TcpStream::connect(socks_proxy)
-        .await
-        .context("Failed to connect to socks proxy")?;
-
-    let socks: Box<dyn AsyncReadWrite> = match version {
         SocksVersion::V4(None) => {
             let socks = Socks4Stream::connect_with_socket(stream, rpc_host)
                 .await
                 .context("error connecting to socks")?;
-            Box::new(socks)
+            Ok(Box::new(socks))
         }
         SocksVersion::V4(Some(Socks4Identification { user_id })) => {
             let socks = Socks4Stream::connect_with_userid_and_socket(stream, rpc_host, user_id)
                 .await
                 .context("error connecting to socks")?;
-            Box::new(socks)
+            Ok(Box::new(socks))
         }
         SocksVersion::V5(None) => {
             let socks = Socks5Stream::connect_with_socket(stream, rpc_host)
                 .await
                 .context("error connecting to socks")?;
-            Box::new(socks)
+            Ok(Box::new(socks))
         }
         SocksVersion::V5(Some(Socks5Authorization { username, password })) => {
             let socks = Socks5Stream::connect_with_password_and_socket(
@@ -64,44 +73,9 @@ pub(crate) async fn connect_socks_proxy_stream(
             )
             .await
             .context("error connecting to socks")?;
-            Box::new(socks)
+            Ok(Box::new(socks))
         }
-    };
+    }
-
-    Ok(socks)
-}
-
-fn parse_socks_proxy(proxy: &Url) -> Option<((String, u16), SocksVersion<'_>)> {
-    let scheme = proxy.scheme();
-    let socks_version = if scheme.starts_with("socks4") {
-        let identification = match proxy.username() {
-            "" => None,
-            username => Some(Socks4Identification { user_id: username }),
-        };
-        SocksVersion::V4(identification)
-    } else if scheme.starts_with("socks") {
-        let authorization = proxy.password().map(|password| Socks5Authorization {
-            username: proxy.username(),
-            password,
-        });
-        SocksVersion::V5(authorization)
-    } else {
-        return None;
-    };
-
-    let host = proxy.host()?.to_string();
-    let port = proxy.port_or_known_default()?;
-
-    Some(((host, port), socks_version))
-}
-
-pub(crate) trait AsyncReadWrite:
-    tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static
-{
-}
-impl<T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static> AsyncReadWrite
-    for T
-{
 }
 
 #[cfg(test)]
@@ -113,20 +87,18 @@ mod tests {
     #[test]
     fn parse_socks4() {
         let proxy = Url::parse("socks4://proxy.example.com:1080").unwrap();
+        let scheme = proxy.scheme();
 
-        let ((host, port), version) = parse_socks_proxy(&proxy).unwrap();
+        let version = parse_socks_proxy(scheme, &proxy);
-        assert_eq!(host, "proxy.example.com");
-        assert_eq!(port, 1080);
         assert!(matches!(version, SocksVersion::V4(None)))
     }
 
     #[test]
     fn parse_socks4_with_identification() {
         let proxy = Url::parse("socks4://userid@proxy.example.com:1080").unwrap();
+        let scheme = proxy.scheme();
 
-        let ((host, port), version) = parse_socks_proxy(&proxy).unwrap();
+        let version = parse_socks_proxy(scheme, &proxy);
-        assert_eq!(host, "proxy.example.com");
-        assert_eq!(port, 1080);
         assert!(matches!(
             version,
             SocksVersion::V4(Some(Socks4Identification { user_id: "userid" }))
@@ -136,20 +108,18 @@ mod tests {
     #[test]
     fn parse_socks5() {
         let proxy = Url::parse("socks5://proxy.example.com:1080").unwrap();
+        let scheme = proxy.scheme();
 
-        let ((host, port), version) = parse_socks_proxy(&proxy).unwrap();
+        let version = parse_socks_proxy(scheme, &proxy);
-        assert_eq!(host, "proxy.example.com");
-        assert_eq!(port, 1080);
         assert!(matches!(version, SocksVersion::V5(None)))
     }
 
     #[test]
     fn parse_socks5_with_authorization() {
         let proxy = Url::parse("socks5://username:password@proxy.example.com:1080").unwrap();
+        let scheme = proxy.scheme();
 
-        let ((host, port), version) = parse_socks_proxy(&proxy).unwrap();
+        let version = parse_socks_proxy(scheme, &proxy);
-        assert_eq!(host, "proxy.example.com");
-        assert_eq!(port, 1080);
         assert!(matches!(
             version,
             SocksVersion::V5(Some(Socks5Authorization {
@@ -158,19 +128,4 @@ mod tests {
             }))
         ))
     }
-
-    /// If parsing the proxy URL fails, we must avoid falling back to an insecure connection.
-    /// SOCKS proxies are often used in contexts where security and privacy are critical,
-    /// so any fallback could expose users to significant risks.
-    #[tokio::test]
-    async fn fails_on_bad_proxy() {
-        // Should fail connecting because http is not a valid Socks proxy scheme
-        let proxy = Url::parse("http://localhost:2313").unwrap();
-
-        let result = connect_socks_proxy_stream(&proxy, ("test", 1080)).await;
-        match result {
-            Err(e) => assert_eq!(e.to_string(), "Parsing proxy url failed"),
-            Ok(_) => panic!("Connecting on bad proxy should fail"),
-        };
-    }
 }