fix: inappropriate connection reuse when using HTTP proxy if the initial CONNECT failed (#2072)

# What This MR Resolves

A CONNECT request is needed to sent to the HTTP proxy first before the
actual client request to establish the tunnel on the proxy. A `HTTP/1.1
200 Connection established` is expected for the initial CONNECT request.
Only when the CONNECT is successful, the client continues sending the
actual request through the "tunnel". And when CONNECT failed, the
connection remains the initial state `unconnected`.

There are following circumstances that a CONNECT fails under but not
limited to following situations:

- The destination is not whitelisted.
- The dest domain can't be resolved(timeout/SERVFAIL/NX/etc.).
- The dest IP can't be connected(timeout/unreachable/etc.).

There could be 2 following strategies to deal with CONNECT failures on
the client side:

1. Close the connection before return to the caller.
2. Mark this connection "unconnected" and put it into the pool. Then
retry the CONNECT next time it's picked out of the pool.

The 2nd one needs to add extra state to Channel in the manager which
brings bigger change to the code.
This MR employs the 1st strategy to resolve it. The issue is described
in #2071 .

# Readings

The CONNECT is documented in `Section 5.3` in RFC2871:
https://www.ietf.org/rfc/rfc2817.txt

The proxy won't actively terminate the connection if the CONNECT failed
if keep-alive is enabled. Unless the tunnel is established and there is
any communication failures in the middle. Therefore the client needs to
deal with this error by its own.

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

Author: jasonjoo2010

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);
+        }
+    }
 }