fix: inappropriate connection reuse when using HTTP proxy if the initial CONNECT failed

client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java

@@ -21,6 +21,7 @@
 import io.netty.handler.codec.DecoderResultProvider;
 import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponse;
 import io.netty.handler.codec.http.LastHttpContent;
@@ -32,6 +33,7 @@
 import org.asynchttpclient.netty.NettyResponseStatus;
 import org.asynchttpclient.netty.channel.ChannelManager;
 import org.asynchttpclient.netty.request.NettyRequestSender;
+import org.asynchttpclient.util.HttpConstants.ResponseStatusCodes;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -43,8 +45,11 @@ public HttpHandler(AsyncHttpClientConfig config, ChannelManager channelManager,
         super(config, channelManager, requestSender);
     }
 
-    private static boolean abortAfterHandlingStatus(AsyncHandler<?> handler, NettyResponseStatus status) throws Exception {
-        return handler.onStatusReceived(status) == State.ABORT;
+    private static boolean abortAfterHandlingStatus(AsyncHandler<?> handler, HttpMethod httpMethod, NettyResponseStatus status) throws Exception {
+        // For non-200 response of a CONNECT request, it's still unconnected.
+        // We need to either close the connection or reuse it but send CONNECT request again.
+        // The former one is easier or we have to attach more state to Channel.
+        return handler.onStatusReceived(status) == State.ABORT || httpMethod == HttpMethod.CONNECT && status.getStatusCode() != ResponseStatusCodes.OK_200;
     }
 
     private static boolean abortAfterHandlingHeaders(AsyncHandler<?> handler, HttpHeaders responseHeaders) throws Exception {
@@ -61,7 +66,7 @@ private void handleHttpResponse(final HttpResponse response, final Channel chann
         HttpHeaders responseHeaders = response.headers();
 
         if (!interceptors.exitAfterIntercept(channel, future, handler, response, status, responseHeaders)) {
-            boolean abort = abortAfterHandlingStatus(handler, status) || abortAfterHandlingHeaders(handler, responseHeaders);
+            boolean abort = abortAfterHandlingStatus(handler, httpRequest.method(), status) || abortAfterHandlingHeaders(handler, responseHeaders);
             if (abort) {
                 finishUpdate(future, channel, true);
             }

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

@@ -13,14 +13,21 @@
 package org.asynchttpclient.proxy;
 
 import io.github.artsok.RepeatedIfExceptionsTest;
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+
 import org.asynchttpclient.AbstractBasicTest;
 import org.asynchttpclient.AsyncHttpClient;
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.RequestBuilder;
 import org.asynchttpclient.Response;
+import org.asynchttpclient.proxy.ProxyServer.Builder;
 import org.asynchttpclient.request.body.generator.ByteArrayBodyGenerator;
 import org.asynchttpclient.test.EchoHandler;
+import org.asynchttpclient.util.HttpConstants;
 import org.eclipse.jetty.proxy.ConnectHandler;
+import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.ServerConnector;
 import org.eclipse.jetty.server.handler.AbstractHandler;
@@ -37,6 +44,8 @@
 import static org.asynchttpclient.test.TestUtils.addHttpsConnector;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
+import java.io.IOException;
+
 /**
  * Proxy usage tests.
  */
@@ -46,7 +55,7 @@ public class HttpsProxyTest extends AbstractBasicTest {
 
     @Override
     public AbstractHandler configureHandler() throws Exception {
-        return new ConnectHandler();
+        return new ProxyHandler();
     }
 
     @Override
@@ -142,4 +151,38 @@ public void testPooledConnectionsWithProxy() throws Exception {
             assertEquals(200, response2.getStatusCode());
         }
     }
+
+    @RepeatedIfExceptionsTest(repeats = 5)
+    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"));
+            RequestBuilder rb = get(getTargetUrl2()).setProxyServer(proxyServer);
+
+            Response response1 = asyncHttpClient.executeRequest(rb.build()).get();
+            assertEquals(403, response1.getStatusCode());
+
+            Response response2 = asyncHttpClient.executeRequest(rb.build()).get();
+            assertEquals(403, response2.getStatusCode());
+
+            Response response3 = asyncHttpClient.executeRequest(rb.build()).get();
+            assertEquals(403, response3.getStatusCode());
+        }
+    }
+
+    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) {
+                    response.setStatus(HttpServletResponse.SC_FORBIDDEN);
+                    r.setHandled(true);
+                    return;
+                }
+            }
+            super.handle(s, r, request, response);
+        }
+    }
 }