Keep connection-authenticated sockets off the HTTP/2 registry

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

@@ -887,6 +887,13 @@ private static void addSslHandlerAtPipelineHead(ChannelPipeline pipeline, String
     }
 
     public Future<Channel> updatePipelineForHttpTunneling(ChannelPipeline pipeline, Uri requestUri) {
+        return updatePipelineForHttpTunneling(pipeline, requestUri, true);
+    }
+
+    /**
+     * @param http2Allowed false when the realm authenticates the socket the tunnel runs over.
+     */
+    public Future<Channel> updatePipelineForHttpTunneling(ChannelPipeline pipeline, Uri requestUri, boolean http2Allowed) {
         Future<Channel> whenHandshaked = null;
 
         if (pipeline.get(HTTP_CLIENT_CODEC) != null) {
@@ -900,7 +907,8 @@ public Future<Channel> updatePipelineForHttpTunneling(ChannelPipeline pipeline,
                 // Remove existing SSL handler (for proxy) and replace with SSL handler for target
                 pipeline.remove(SSL_HANDLER);
             }
-            SslHandler sslHandler = createSslHandler(requestUri.getHost(), requestUri.getExplicitPort(), !requestUri.isWebSocket());
+            SslHandler sslHandler = createSslHandler(requestUri.getHost(), requestUri.getExplicitPort(),
+                    http2Allowed && !requestUri.isWebSocket());
             whenHandshaked = sslHandler.handshakeFuture();
             addSslHandlerAtPipelineHead(pipeline, SSL_HANDLER, sslHandler);
             pipeline.addAfter(SSL_HANDLER, HTTP_CLIENT_CODEC, newHttpClientCodec());
@@ -926,6 +934,13 @@ public Future<Channel> updatePipelineForHttpTunneling(ChannelPipeline pipeline,
     }
 
     public Future<Channel> updatePipelineForHttpsTunneling(ChannelPipeline pipeline, Uri requestUri, ProxyServer proxyServer) {
+        return updatePipelineForHttpsTunneling(pipeline, requestUri, true);
+    }
+
+    /**
+     * @param http2Allowed false when the realm authenticates the socket the tunnel runs over.
+     */
+    public Future<Channel> updatePipelineForHttpsTunneling(ChannelPipeline pipeline, Uri requestUri, boolean http2Allowed) {
         Future<Channel> whenHandshaked = null;
 
         // Remove HTTP codec as tunnel is established
@@ -938,7 +953,8 @@ public Future<Channel> updatePipelineForHttpsTunneling(ChannelPipeline pipeline,
             // The proxy SSL handler should remain as it provides the tunnel transport
             // We need to add target SSL handler that will negotiate with the target through the tunnel
 
-            SslHandler sslHandler = createSslHandler(requestUri.getHost(), requestUri.getExplicitPort(), !requestUri.isWebSocket());
+            SslHandler sslHandler = createSslHandler(requestUri.getHost(), requestUri.getExplicitPort(),
+                    http2Allowed && !requestUri.isWebSocket());
             whenHandshaked = sslHandler.handshakeFuture();
 
             // For HTTPS proxy tunnel, add target SSL handler after the existing proxy SSL handler
@@ -975,6 +991,15 @@ public Future<Channel> updatePipelineForHttpsTunneling(ChannelPipeline pipeline,
     }
 
     public SslHandler addSslHandler(ChannelPipeline pipeline, Uri uri, String virtualHost, boolean hasSocksProxyHandler) {
+        return addSslHandler(pipeline, uri, virtualHost, hasSocksProxyHandler, true);
+    }
+
+    /**
+     * @param http2Allowed false for a WebSocket (no RFC 8441 support) and for a realm that authenticates the
+     *                     socket, since every stream of an h2 connection shares it.
+     */
+    public SslHandler addSslHandler(ChannelPipeline pipeline, Uri uri, String virtualHost, boolean hasSocksProxyHandler,
+                                    boolean http2Allowed) {
         String peerHost;
         int peerPort;
 
@@ -994,7 +1019,7 @@ public SslHandler addSslHandler(ChannelPipeline pipeline, Uri uri, String virtua
         }
 
         // A WebSocket connection must not negotiate h2 (no RFC 8441 support), so advertise only http/1.1 in ALPN.
-        SslHandler sslHandler = createSslHandler(peerHost, peerPort, !uri.isWebSocket());
+        SslHandler sslHandler = createSslHandler(peerHost, peerPort, http2Allowed && !uri.isWebSocket());
         // Check if SOCKS handler actually exists in the pipeline before trying to add after it
         if (hasSocksProxyHandler && pipeline.get(SOCKS_HANDLER) != null) {
             pipeline.addAfter(SOCKS_HANDLER, SSL_HANDLER, sslHandler);
@@ -1257,14 +1282,33 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
      * tunneled request is sent.
      */
     public void upgradePipelineToHttp2AfterProxyConnect(ChannelPipeline pipeline, Object partitionKey) {
+        upgradePipelineToHttp2AfterProxyConnect(pipeline, partitionKey, true);
+    }
+
+    /**
+     * @param shareable false when the tunnel is authenticated, so no other principal can draw it.
+     */
+    public void upgradePipelineToHttp2AfterProxyConnect(ChannelPipeline pipeline, Object partitionKey, boolean shareable) {
         SslHandler targetSslHandler = (SslHandler) pipeline.get(TARGET_SSL_HANDLER);
         if (targetSslHandler == null) {
             targetSslHandler = (SslHandler) pipeline.get(SSL_HANDLER);
         }
         if (targetSslHandler != null
                 && ApplicationProtocolNames.HTTP_2.equals(targetSslHandler.applicationProtocol())) {
             upgradePipelineToHttp2(pipeline);
-            registerHttp2Connection(partitionKey, pipeline.channel());
+            if (shareable) {
+                registerHttp2Connection(partitionKey, pipeline.channel());
+            } else {
+                // Unregistered, and an h2 parent is never offered to the HTTP/1.1 pool either, so it would
+                // linger holding its permit.
+                Http2ConnectionState state = pipeline.channel().attr(Http2ConnectionState.HTTP2_STATE_KEY).get();
+                if (state != null) {
+                    state.markRedundant();
+                }
+                // The peer does speak h2, so don't mark the host as non-HTTP/2; just release whatever was
+                // waiting for the connection this would have registered.
+                wakeHttp2ConnectionWaiters(partitionKey, null);
+            }
         } else if (targetSslHandler != null) {
             http2Unavailable(partitionKey);
         }

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

@@ -196,7 +196,8 @@ protected void onFailure(Throwable cause) {
         } else if ((proxyServer == null || proxyServer.getProxyType().isSocks()) && uri.isSecured()) {
             SslHandler sslHandler;
             try {
-                sslHandler = channelManager.addSslHandler(channel.pipeline(), uri, request.getVirtualHost(), proxyServer != null);
+                sslHandler = channelManager.addSslHandler(channel.pipeline(), uri, request.getVirtualHost(), proxyServer != null,
+                        !PrincipalScopedPartitionKey.authenticatesTheConnection(future.getRealm(), future.getProxyRealm()));
             } catch (Exception sslError) {
                 onFailure(channel, sslError);
                 return;
@@ -271,7 +272,19 @@ protected void onFailure(Throwable cause) {
         } else {
             // h2c (cleartext HTTP/2 prior knowledge): upgrade to HTTP/2 without TLS. WebSocket (ws://) is
             // excluded for the same RFC 8441 reason as the TLS path above — it stays on HTTP/1.1.
-            if (!uri.isSecured() && channelManager.isHttp2CleartextEnabled() && !uri.isWebSocket()) {
+            // Upgrading and then withholding it from the registry would only orphan the socket, and the
+            // 401 retry needs the one that issued the challenge. The HTTP/1.1 pool is scoped by principal
+            // for the origin hop only, so for a proxy realm this bounds the sharing rather than ending it.
+            boolean connectionAuth = PrincipalScopedPartitionKey.authenticatesTheConnection(future.getRealm(),
+                    future.getProxyRealm());
+            if (connectionAuth && !uri.isSecured() && channelManager.isHttp2CleartextEnabled() && !uri.isWebSocket()) {
+                // h2c is prior knowledge (RFC 9113 3.3): there is no negotiation to carry the refusal, so a
+                // server that speaks only h2 sees an HTTP/1.1 request and answers with a preface mismatch.
+                LOGGER.warn("Sending {} over HTTP/1.1 instead of h2c: {} authenticates the connection. "
+                        + "Disable http2Cleartext for this client if the server speaks only HTTP/2.",
+                        uri, future.getRealm() != null ? future.getRealm().getScheme() : future.getProxyRealm().getScheme());
+            }
+            if (!uri.isSecured() && channelManager.isHttp2CleartextEnabled() && !uri.isWebSocket() && !connectionAuth) {
                 try {
                     channelManager.upgradePipelineToHttp2(channel.pipeline());
                 } catch (Exception upgradeError) {
@@ -314,6 +327,15 @@ private void registerHttp2AndManageSemaphore(Channel channel, ConnectionSemaphor
         // pool is polled with, including the IP-aware key used by LoadBalance.ROUND_ROBIN. Read the key
         // once: it is volatile (repinned on IP failover) and both uses below must agree.
         Object partitionKey = future.getPartitionKey();
+        if (PrincipalScopedPartitionKey.authenticatesTheConnection(future.getRealm(), future.getProxyRealm())) {
+            // Only reachable when a caller-supplied SslEngineFactory advertised h2 anyway. It cannot serve
+            // HTTP/1.1 now, so let it close after this exchange; returning bare would strand the permit.
+            Http2ConnectionState state = channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get();
+            if (state != null) {
+                state.markRedundant();
+            }
+            return;
+        }
         channelManager.registerHttp2Connection(partitionKey, channel);
         if (partitionKey instanceof RoundRobinPartitionKey) {
             // The permit must be freed as soon as the connection stops serving new requests: on close, or

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

@@ -16,6 +16,7 @@
 package org.asynchttpclient.netty.channel;
 
 import org.asynchttpclient.Realm;
+import org.jetbrains.annotations.Nullable;
 
 import java.util.Objects;
 
@@ -59,6 +60,22 @@ public static Object scope(Object baseKey, Realm realm) {
         return new PrincipalScopedPartitionKey(baseKey, realm.getScheme(), realm.getPrincipal());
     }
 
+    /**
+     * Whether the scheme authenticates the connection rather than the request. Unlike
+     * {@link #scope(Object, Realm)} this needs no principal: an ambient Kerberos login carries none and still
+     * binds the socket.
+     */
+    public static boolean authenticatesTheConnection(@Nullable Realm realm) {
+        return realm != null && authenticatesTheConnection(realm.getScheme());
+    }
+
+    /**
+     * Either hop, since a CONNECT tunnel is one socket.
+     */
+    public static boolean authenticatesTheConnection(@Nullable Realm realm, @Nullable Realm proxyRealm) {
+        return authenticatesTheConnection(realm) || authenticatesTheConnection(proxyRealm);
+    }
+
     private static boolean authenticatesTheConnection(Realm.AuthScheme scheme) {
         return scheme == Realm.AuthScheme.NTLM
                 || scheme == Realm.AuthScheme.KERBEROS

client/src/main/java/org/asynchttpclient/netty/handler/intercept/ConnectSuccessInterceptor.java

@@ -21,6 +21,7 @@
 import org.asynchttpclient.Request;
 import org.asynchttpclient.netty.NettyResponseFuture;
 import org.asynchttpclient.netty.channel.ChannelManager;
+import org.asynchttpclient.netty.channel.PrincipalScopedPartitionKey;
 import org.asynchttpclient.netty.request.NettyRequestSender;
 import org.asynchttpclient.proxy.ProxyServer;
 import org.asynchttpclient.proxy.ProxyType;
@@ -56,13 +57,15 @@ public boolean exitAfterHandlingConnect(Channel channel, NettyResponseFuture<?>
 
         final Future<Channel> whenHandshaked;
 
+        boolean http2Allowed = !PrincipalScopedPartitionKey.authenticatesTheConnection(future.getRealm(), future.getProxyRealm());
+
         // Special handling for HTTPS proxy tunneling
         if (proxyServer != null && ProxyType.HTTPS.equals(proxyServer.getProxyType())) {
             // For HTTPS proxy, we need special tunnel pipeline management
-            whenHandshaked = channelManager.updatePipelineForHttpsTunneling(channel.pipeline(), requestUri, proxyServer);
+            whenHandshaked = channelManager.updatePipelineForHttpsTunneling(channel.pipeline(), requestUri, http2Allowed);
         } else {
             // Standard HTTP proxy or SOCKS proxy tunneling
-            whenHandshaked = channelManager.updatePipelineForHttpTunneling(channel.pipeline(), requestUri);
+            whenHandshaked = channelManager.updatePipelineForHttpTunneling(channel.pipeline(), requestUri, http2Allowed);
         }
 
         future.setReuseChannel(true);

client/src/main/java/org/asynchttpclient/netty/handler/intercept/Redirect30xInterceptor.java

@@ -227,6 +227,11 @@ public boolean exitAfterHandlingRedirect(Channel channel, NettyResponseFuture<?>
                     requestBuilder.setRefuseCrossOriginBodyOnRedirect(refuseCrossOriginBody);
                 }
 
+                // Sampled before stripAuth clears the realm: the channel being drained is authenticated to
+                // the previous origin, and reading it afterwards would file it unscoped.
+                final Object initialPartitionKey = PrincipalScopedPartitionKey.scope(
+                        future.getPartitionKey(), future.getRealm());
+
                 if (stripAuth) {
                     future.setRealm(null);
                     future.setProxyRealm(null);
@@ -237,11 +242,6 @@ public boolean exitAfterHandlingRedirect(Channel channel, NettyResponseFuture<?>
                 // in case of a redirect from HTTP to HTTPS, future
                 // attributes might change
                 final boolean initialConnectionKeepAlive = future.isKeepAlive();
-                // Scoped like every other offer: an NTLM or Negotiate connection must stay with the
-                // identity that authenticated it, and a same-host redirect during NTLM is the ordinary
-                // case rather than an exotic one.
-                final Object initialPartitionKey = PrincipalScopedPartitionKey.scope(
-                        future.getPartitionKey(), future.getRealm());
 
                 CookieStore cookieStore = config.getCookieStore();
                 if (cookieStore != null) {

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

@@ -1313,6 +1313,11 @@ private <T> ListenableFuture<T> reuseOrDeferHttp2Connection(Request request, Pro
         if (request.getUri().isWebSocket()) {
             return null;
         }
+        // Nor a request whose realm authenticates the connection: see registerHttp2AndManageSemaphore.
+        if (PrincipalScopedPartitionKey.authenticatesTheConnection(pooledIdentity(future, request),
+                pooledProxyIdentity(future, proxy))) {
+            return null;
+        }
         // In round-robin mode, only multiplex onto the H2 connection for the IP this request is pinned to;
         // otherwise use the per-host base key. Derive it from the live request rather than
         // future.basePartitionKey() so a filter replay that rewrites the host still polls the correct key
@@ -1469,6 +1474,16 @@ private boolean isOnEventLoop() {
         return false;
     }
 
+    /**
+     * Resolved as newNettyRequestAndResponseFuture resolves it, so the poll and the offer agree.
+     */
+    private @Nullable Realm pooledProxyIdentity(@Nullable NettyResponseFuture<?> future, @Nullable ProxyServer proxy) {
+        if (future != null) {
+            return future.getProxyRealm();
+        }
+        return proxy != null ? proxy.getRealm() : null;
+    }
+
     /**
      * The realm whose identity a pooled connection is scoped by. Resolved exactly as
      * newNettyRequestAndResponseFuture resolves it, so the poll and the offer agree. The future is null on
@@ -1499,8 +1514,10 @@ private Channel pollPooledChannel(NettyResponseFuture<?> future, Request request
         // Round-robin mode: poll with the IP-aware key so reuse stays pinned to the chosen IP (both the
         // HTTP/2 registry and the HTTP/1.1 pool).
         Object override = future != null ? future.getPartitionKeyOverride() : null;
+        boolean http2Shareable = !PrincipalScopedPartitionKey.authenticatesTheConnection(
+                pooledIdentity(future, request), pooledProxyIdentity(future, proxy));
         if (override != null) {
-            if (!uri.isWebSocket()) {
+            if (!uri.isWebSocket() && http2Shareable) {
                 Channel h2Channel = channelManager.pollHttp2Connection(override);
                 if (h2Channel != null) {
                     if (LOGGER.isDebugEnabled()) {
@@ -1531,7 +1548,7 @@ private Channel pollPooledChannel(NettyResponseFuture<?> future, Request request
         // (replayRequest) the future's targetRequest is not updated to the replayed request, so its memoized
         // base key can lag a host-rewriting replay. Reading the current request's URI/virtualHost stays correct.
         Object partitionKey = request.getChannelPoolPartitioning().getPartitionKey(uri, virtualHost, proxy);
-        if (!uri.isWebSocket()) {
+        if (!uri.isWebSocket() && http2Shareable) {
             Channel h2Channel = channelManager.pollHttp2Connection(partitionKey);
             if (h2Channel != null) {
                 if (LOGGER.isDebugEnabled()) {
@@ -1644,7 +1661,9 @@ public void call() {
                                 if (!nextRequest.getUri().isWebSocket()) {
                                     try {
                                         channelManager.upgradePipelineToHttp2AfterProxyConnect(
-                                                channel.pipeline(), future.getPartitionKey());
+                                                channel.pipeline(), future.getPartitionKey(),
+                                                !PrincipalScopedPartitionKey.authenticatesTheConnection(
+                                                        future.getRealm(), future.getProxyRealm()));
                                     } catch (Exception upgradeError) {
                                         // Thrown inside a future listener, which logs and discards it.
                                         // Fail the future instead, or the request waits for its timeout.