fix: send CONNECT first when recovering a HTTPS request

Signed-off-by: Jason Joo <[email protected]>

Author: jasonjoo2010

client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java

@@ -97,6 +97,13 @@ public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelMa
         requestFactory = new NettyRequestFactory(config);
     }
 
+    // needConnect returns true if the request is secure/websocket and a HTTP proxy is set
+    private boolean needConnect(final Request request, final ProxyServer proxyServer) {
+        return proxyServer != null
+                && proxyServer.getProxyType().isHttp()
+                && (request.getUri().isSecured() || request.getUri().isWebSocket());
+    }
+
     public <T> ListenableFuture<T> sendRequest(final Request request, final AsyncHandler<T> asyncHandler, NettyResponseFuture<T> future) {
         if (isClosed()) {
             throw new IllegalStateException("Closed");
@@ -106,9 +113,7 @@ public <T> ListenableFuture<T> sendRequest(final Request request, final AsyncHan
         ProxyServer proxyServer = getProxyServer(config, request);
 
         // WebSockets use connect tunneling to work with proxies
-        if (proxyServer != null && proxyServer.getProxyType().isHttp() &&
-                (request.getUri().isSecured() || request.getUri().isWebSocket()) &&
-                !isConnectAlreadyDone(request, future)) {
+        if (needConnect(request, proxyServer) && !isConnectAlreadyDone(request, future)) {
             // Proxy with HTTPS or WebSocket: CONNECT for sure
             if (future != null && future.isConnectAllowed()) {
                 // Perform CONNECT
@@ -125,6 +130,8 @@ public <T> ListenableFuture<T> sendRequest(final Request request, final AsyncHan
 
     private static boolean isConnectAlreadyDone(Request request, NettyResponseFuture<?> future) {
         return future != null
+                // If the channel can't be reused or closed, a CONNECT is still required
+                && future.isReuseChannel() && Channels.isChannelActive(future.channel())
                 && future.getNettyRequest() != null
                 && future.getNettyRequest().getHttpRequest().method() == HttpMethod.CONNECT
                 && !request.getMethod().equals(CONNECT);
@@ -137,11 +144,19 @@ private static boolean isConnectAlreadyDone(Request request, NettyResponseFuture
      */
     private <T> ListenableFuture<T> sendRequestWithCertainForceConnect(Request request, AsyncHandler<T> asyncHandler, NettyResponseFuture<T> future,
                                                                        ProxyServer proxyServer, boolean performConnectRequest) {
-        NettyResponseFuture<T> newFuture = newNettyRequestAndResponseFuture(request, asyncHandler, future, proxyServer, performConnectRequest);
         Channel channel = getOpenChannel(future, request, proxyServer, asyncHandler);
-        return Channels.isChannelActive(channel)
-                ? sendRequestWithOpenChannel(newFuture, asyncHandler, channel)
-                : sendRequestWithNewChannel(request, proxyServer, newFuture, asyncHandler);
+        if (Channels.isChannelActive(channel)) {
+            NettyResponseFuture<T> newFuture = newNettyRequestAndResponseFuture(request, asyncHandler, future,
+                    proxyServer, performConnectRequest);
+            return sendRequestWithOpenChannel(newFuture, asyncHandler, channel);
+        } else {
+            // A new channel is not expected when performConnectRequest is false. We need to
+            // revisit the condition of sending
+            // the CONNECT request to the new channel.
+            NettyResponseFuture<T> newFuture = newNettyRequestAndResponseFuture(request, asyncHandler, future,
+                    proxyServer, needConnect(request, proxyServer));
+            return sendRequestWithNewChannel(request, proxyServer, newFuture, asyncHandler);
+        }
     }
 
     /**

client/src/test/java/org/asynchttpclient/proxy/HttpsProxyTest.java

@@ -13,6 +13,7 @@
 package org.asynchttpclient.proxy;
 
 import io.github.artsok.RepeatedIfExceptionsTest;
+import io.netty.handler.codec.http.DefaultHttpHeaders;
 import jakarta.servlet.ServletException;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
@@ -43,8 +44,10 @@
 import static org.asynchttpclient.test.TestUtils.addHttpConnector;
 import static org.asynchttpclient.test.TestUtils.addHttpsConnector;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
 
 import java.io.IOException;
+import java.util.concurrent.ExecutionException;
 
 /**
  * Proxy usage tests.
@@ -156,7 +159,7 @@ public void testPooledConnectionsWithProxy() throws Exception {
     public void testFailedConnectWithProxy() throws Exception {
         try (AsyncHttpClient asyncHttpClient = asyncHttpClient(config().setFollowRedirect(true).setUseInsecureTrustManager(true).setKeepAlive(true))) {
         	Builder proxyServer = proxyServer("localhost", port1);
-        	proxyServer.setCustomHeaders(r -> r.getHeaders().add(ProxyHandler.HEADER_FORBIDDEN, "1"));
+        	proxyServer.setCustomHeaders(r -> new DefaultHttpHeaders().set(ProxyHandler.HEADER_FORBIDDEN, "1"));
             RequestBuilder rb = get(getTargetUrl2()).setProxyServer(proxyServer);
 
             Response response1 = asyncHttpClient.executeRequest(rb.build()).get();
@@ -170,16 +173,39 @@ public void testFailedConnectWithProxy() throws Exception {
         }
     }
 
+    @RepeatedIfExceptionsTest(repeats = 5)
+    public void testClosedConnectionWithProxy() throws Exception {
+        try (AsyncHttpClient asyncHttpClient = asyncHttpClient(
+                config().setFollowRedirect(true).setUseInsecureTrustManager(true).setKeepAlive(true))) {
+            Builder proxyServer = proxyServer("localhost", port1);
+            proxyServer.setCustomHeaders(r -> new DefaultHttpHeaders().set(ProxyHandler.HEADER_FORBIDDEN, "2"));
+            RequestBuilder rb = get(getTargetUrl2()).setProxyServer(proxyServer);
+
+            assertThrowsExactly(ExecutionException.class, () -> asyncHttpClient.executeRequest(rb.build()).get());
+            assertThrowsExactly(ExecutionException.class, () -> asyncHttpClient.executeRequest(rb.build()).get());
+            assertThrowsExactly(ExecutionException.class, () -> asyncHttpClient.executeRequest(rb.build()).get());
+        }
+    }
+
     public static class ProxyHandler extends ConnectHandler {
     	final static String HEADER_FORBIDDEN = "X-REJECT-REQUEST";
 
         @Override
         public void handle(String s, Request r, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
             if (HttpConstants.Methods.CONNECT.equalsIgnoreCase(request.getMethod())) {
-                if (request.getHeader(HEADER_FORBIDDEN) != null) {
+                String headerValue = request.getHeader(HEADER_FORBIDDEN);
+                if (headerValue == null) {
+                    headerValue = "";
+                }
+                switch (headerValue) {
+                case "1":
                     response.setStatus(HttpServletResponse.SC_FORBIDDEN);
                     r.setHandled(true);
                     return;
+                case "2":
+                    r.getHttpChannel().getConnection().close();
+                    r.setHandled(true);
+                    return;
                 }
             }
             super.handle(s, r, request, response);