fix(ext/fetch): use correct ALPN to socks5 proxies (#24817)

Closes #24632 
Closes #24719

Cargo.lock

@@ -1478,6 +1478,7 @@ dependencies = [
  "deno_permissions",
  "deno_tls",
  "dyn-clone",
+ "fast-socks5",
  "http 1.1.0",
  "http-body-util",
  "hyper 1.4.1",
@@ -2693,6 +2694,20 @@ dependencies = [
  "regex",
 ]
 
+[[package]]
+name = "fast-socks5"
+version = "0.9.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f89f36d4ee12370d30d57b16c7e190950a1a916e7dbbb5fd5a412f5ef913fe84"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "log",
+ "thiserror",
+ "tokio",
+ "tokio-stream",
+]
+
 [[package]]
 name = "faster-hex"
 version = "0.9.0"

Cargo.toml

@@ -108,6 +108,7 @@ dlopen2 = "0.6.1"
 ecb = "=0.1.2"
 elliptic-curve = { version = "0.13.4", features = ["alloc", "arithmetic", "ecdh", "std", "pem"] }
 encoding_rs = "=0.8.33"
+fast-socks5 = "0.9.6"
 faster-hex = "0.9"
 fastwebsockets = { version = "0.6", features = ["upgrade", "unstable-split"] }
 filetime = "0.2.16"

ext/fetch/Cargo.toml

@@ -38,3 +38,6 @@ tokio-util = { workspace = true, features = ["io"] }
 tower.workspace = true
 tower-http.workspace = true
 tower-service.workspace = true
+
+[dev-dependencies]
+fast-socks5.workspace = true

ext/fetch/proxy.rs

@@ -727,7 +727,14 @@ where
         }
       }
       Proxied::Socks(ref p) => p.connected(),
-      Proxied::SocksTls(ref p) => p.inner().get_ref().0.connected(),
+      Proxied::SocksTls(ref p) => {
+        let tunneled_tls = p.inner().get_ref();
+        if tunneled_tls.1.alpn_protocol() == Some(b"h2") {
+          tunneled_tls.0.connected().negotiated_h2()
+        } else {
+          tunneled_tls.0.connected()
+        }
+      }
     }
   }
 }

ext/fetch/tests.rs

@@ -4,6 +4,8 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 
 use bytes::Bytes;
+use fast_socks5::server::Config as Socks5Config;
+use fast_socks5::server::Socks5Socket;
 use http_body_util::BodyExt;
 use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
@@ -19,27 +21,41 @@ static EXAMPLE_KEY: &[u8] =
 async fn test_https_proxy_http11() {
   let src_addr = create_https_server(false).await;
   let prx_addr = create_http_proxy(src_addr).await;
-  run_test_client(prx_addr, src_addr, false, http::Version::HTTP_11).await;
+  run_test_client(prx_addr, src_addr, "http", http::Version::HTTP_11).await;
 }
 
 #[tokio::test]
 async fn test_https_proxy_h2() {
   let src_addr = create_https_server(true).await;
   let prx_addr = create_http_proxy(src_addr).await;
-  run_test_client(prx_addr, src_addr, false, http::Version::HTTP_2).await;
+  run_test_client(prx_addr, src_addr, "http", http::Version::HTTP_2).await;
 }
 
 #[tokio::test]
 async fn test_https_proxy_https_h2() {
   let src_addr = create_https_server(true).await;
   let prx_addr = create_https_proxy(src_addr).await;
-  run_test_client(prx_addr, src_addr, true, http::Version::HTTP_2).await;
+  run_test_client(prx_addr, src_addr, "https", http::Version::HTTP_2).await;
+}
+
+#[tokio::test]
+async fn test_socks_proxy_http11() {
+  let src_addr = create_https_server(false).await;
+  let prx_addr = create_socks_proxy(src_addr).await;
+  run_test_client(prx_addr, src_addr, "socks5", http::Version::HTTP_11).await;
+}
+
+#[tokio::test]
+async fn test_socks_proxy_h2() {
+  let src_addr = create_https_server(true).await;
+  let prx_addr = create_socks_proxy(src_addr).await;
+  run_test_client(prx_addr, src_addr, "socks5", http::Version::HTTP_2).await;
 }
 
 async fn run_test_client(
   prx_addr: SocketAddr,
   src_addr: SocketAddr,
-  https: bool,
+  proto: &str,
   ver: http::Version,
 ) {
   let client = create_http_client(
@@ -48,7 +64,7 @@ async fn run_test_client(
       root_cert_store: None,
       ca_certs: vec![],
       proxy: Some(deno_tls::Proxy {
-        url: format!("http{}://{}", if https { "s" } else { "" }, prx_addr),
+        url: format!("{}://{}", proto, prx_addr),
         basic_auth: None,
       }),
       unsafely_ignore_certificate_errors: Some(vec![]),
@@ -186,3 +202,29 @@ async fn create_https_proxy(src_addr: SocketAddr) -> SocketAddr {
 
   prx_addr
 }
+
+async fn create_socks_proxy(src_addr: SocketAddr) -> SocketAddr {
+  let prx_tcp = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+  let prx_addr = prx_tcp.local_addr().unwrap();
+
+  tokio::spawn(async move {
+    while let Ok((sock, _)) = prx_tcp.accept().await {
+      let cfg: Socks5Config = Default::default();
+      let mut socks_conn = Socks5Socket::new(sock, cfg.into())
+        .upgrade_to_socks5()
+        .await
+        .unwrap();
+
+      let fut = async move {
+        let mut dst_tcp =
+          tokio::net::TcpStream::connect(src_addr).await.unwrap();
+        tokio::io::copy_bidirectional(&mut socks_conn, &mut dst_tcp)
+          .await
+          .unwrap();
+      };
+      tokio::spawn(fut);
+    }
+  });
+
+  prx_addr
+}