Never pool a socket whose CONNECT the proxy refused

Motivation:
A refused CONNECT leaves a plaintext hop to the proxy. The status check in
HttpHandler closes it on the ordinary path, but response filters run ahead of
that check, so a ResponseFilter asking for a replay - an everyday retry
pattern - short-circuits the interceptor chain and the socket reaches the
pool. The next exchange for that origin then polls it back and writes the
origin request, Authorization included, down a hop the proxy is still reading
in the clear. The 2.x line already guards this at the pool boundary; 3.x did
not.

Modification:
Refuse to pool a channel whose exchange still carries a CONNECT. A CONNECT
the proxy accepted is consumed by ConnectSuccessInterceptor, which takes the
channel over, so a CONNECT still on the future here means the tunnel was
never built. Sampled where keepAlive and the partition key are sampled,
because callers hand the drain over and immediately move the future on to the
next request.

Result:
A refused-CONNECT socket is closed rather than pooled, on the filter-replay
path as well as the ordinary one, matching 2.16.1.

Author: hyperxpro

client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -32,6 +32,7 @@
 import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.websocketx.WebSocket08FrameDecoder;
 import io.netty.handler.codec.http.websocketx.WebSocket08FrameEncoder;
 import io.netty.handler.codec.http.websocketx.WebSocketFrameAggregator;
@@ -81,6 +82,7 @@
 import org.asynchttpclient.netty.handler.Http2PingHandler;
 import org.asynchttpclient.netty.handler.HttpHandler;
 import org.asynchttpclient.netty.handler.WebSocketHandler;
+import org.asynchttpclient.netty.request.NettyRequest;
 import org.asynchttpclient.netty.request.NettyRequestSender;
 import org.asynchttpclient.netty.ssl.DefaultSslEngineFactory;
 import org.asynchttpclient.proxy.ProxyServer;
@@ -447,6 +449,36 @@ private Http1ContentDecompressor newHttpContentDecompressor() {
         return new Http1ContentDecompressor(config.isKeepEncodingHeader(), config.getMaxDecompressedResponseSize());
     }
 
+    /**
+     * Whether the exchange being finished was a CONNECT that never became a tunnel. A CONNECT the proxy
+     * accepted is consumed by {@link org.asynchttpclient.netty.handler.intercept.ConnectSuccessInterceptor},
+     * which takes the channel over, so a CONNECT still on the future here means the proxy refused it and the
+     * socket is a plaintext hop to the proxy. Pooling it would let the next exchange for that origin send the
+     * origin request, {@code Authorization} included, down a hop the proxy is still reading in the clear.
+     * <p>
+     * The status check in {@code HttpHandler} closes this on the ordinary path, but a {@code ResponseFilter}
+     * asking for a replay short-circuits {@code exitAfterIntercept} before that check runs, so the guard has
+     * to live at the pool boundary too.
+     */
+    private static boolean isRefusedTunnel(NettyResponseFuture<?> future) {
+        NettyRequest nettyRequest = future.getNettyRequest();
+        return nettyRequest != null && nettyRequest.getHttpRequest().method() == HttpMethod.CONNECT;
+    }
+
+    public void tryToOfferChannelToPool(Channel channel, NettyResponseFuture<?> future, boolean keepAlive, Object partitionKey) {
+        tryToOfferChannelToPool(channel, future.getAsyncHandler(), keepAlive, partitionKey, isRefusedTunnel(future));
+    }
+
+    private void tryToOfferChannelToPool(Channel channel, AsyncHandler<?> asyncHandler, boolean keepAlive, Object partitionKey,
+                                         boolean refusedTunnel) {
+        if (refusedTunnel) {
+            LOGGER.debug("Not offering channel {} to the pool: the CONNECT on it was never established", channel);
+            closeChannel(channel);
+            return;
+        }
+        tryToOfferChannelToPool(channel, asyncHandler, keepAlive, partitionKey);
+    }
+
     public final void tryToOfferChannelToPool(Channel channel, AsyncHandler<?> asyncHandler, boolean keepAlive, Object partitionKey) {
         if (channel.isActive() && keepAlive) {
             LOGGER.debug("Adding key: {} for channel {}", partitionKey, channel);
@@ -1172,10 +1204,16 @@ public void upgradePipelineForWebSockets(ChannelPipeline pipeline) {
     }
 
     private OnLastHttpContentCallback newDrainCallback(final NettyResponseFuture<?> future, final Channel channel, final boolean keepAlive, final Object partitionKey) {
+
+        // Sampled here rather than in call(), for the same reason keepAlive and partitionKey are: callers
+        // hand the drain over and immediately move the future on to the NEXT request, so by the time the
+        // last chunk arrives the future no longer describes the response being drained off this channel.
+        final boolean refusedTunnel = isRefusedTunnel(future);
+
         return new OnLastHttpContentCallback(future) {
             @Override
             public void call() {
-                tryToOfferChannelToPool(channel, future.getAsyncHandler(), keepAlive, partitionKey);
+                tryToOfferChannelToPool(channel, future.getAsyncHandler(), keepAlive, partitionKey, refusedTunnel);
             }
         };
     }

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

@@ -200,7 +200,7 @@ void finishUpdate(NettyResponseFuture<?> future, Channel channel, boolean close)
         if (close) {
             channelManager.closeChannel(channel);
         } else {
-            channelManager.tryToOfferChannelToPool(channel, future.getAsyncHandler(), true, future.getPartitionKey());
+            channelManager.tryToOfferChannelToPool(channel, future, true, future.getPartitionKey());
         }
 
         try {