From c98b25f3000b58f5de7f0e3644070b9656c0d1a8 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Sun, 20 Sep 2026 18:09:40 +0000 Subject: [PATCH 1/2] Keep connection-authenticated sockets off the HTTP/2 registry --- .../netty/channel/ChannelManager.java | 53 ++++- .../netty/channel/NettyConnectListener.java | 26 ++- .../channel/PrincipalScopedPartitionKey.java | 17 ++ .../intercept/ConnectSuccessInterceptor.java | 8 +- .../intercept/Redirect30xInterceptor.java | 10 +- .../netty/request/NettyRequestSender.java | 25 +- .../ConnectionOrientedAuthHttp2Test.java | 139 +++++++++++ ...ectionOrientedAuthConnectListenerTest.java | 220 ++++++++++++++++++ .../ConnectionOrientedAuthPollGateTest.java | 172 ++++++++++++++ 9 files changed, 654 insertions(+), 16 deletions(-) create mode 100644 client/src/test/java/org/asynchttpclient/netty/channel/ConnectionOrientedAuthHttp2Test.java create mode 100644 client/src/test/java/org/asynchttpclient/netty/request/ConnectionOrientedAuthConnectListenerTest.java create mode 100644 client/src/test/java/org/asynchttpclient/netty/request/ConnectionOrientedAuthPollGateTest.java diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index 4b4d0f83f..f9c6e8d4b 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -887,6 +887,13 @@ private static void addSslHandlerAtPipelineHead(ChannelPipeline pipeline, String } public Future 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 updatePipelineForHttpTunneling(ChannelPipeline pipeline, Uri requestUri, boolean http2Allowed) { Future whenHandshaked = null; if (pipeline.get(HTTP_CLIENT_CODEC) != null) { @@ -900,7 +907,8 @@ public Future 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,14 @@ public Future updatePipelineForHttpTunneling(ChannelPipeline pipeline, } public Future updatePipelineForHttpsTunneling(ChannelPipeline pipeline, Uri requestUri, ProxyServer proxyServer) { + return updatePipelineForHttpsTunneling(pipeline, requestUri, proxyServer, true); + } + + /** + * @param http2Allowed false when the realm authenticates the socket the tunnel runs over. + */ + public Future updatePipelineForHttpsTunneling(ChannelPipeline pipeline, Uri requestUri, ProxyServer proxyServer, + boolean http2Allowed) { Future whenHandshaked = null; // Remove HTTP codec as tunnel is established @@ -938,7 +954,8 @@ public Future 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 +992,15 @@ public Future 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 +1020,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,6 +1283,13 @@ 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); @@ -1264,7 +1297,19 @@ public void upgradePipelineToHttp2AfterProxyConnect(ChannelPipeline pipeline, Ob 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); } diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java b/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java index 5f7814657..93cfc7018 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java +++ b/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 diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/PrincipalScopedPartitionKey.java b/client/src/main/java/org/asynchttpclient/netty/channel/PrincipalScopedPartitionKey.java index b275ae120..88002b06a 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/PrincipalScopedPartitionKey.java +++ b/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 diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ConnectSuccessInterceptor.java b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ConnectSuccessInterceptor.java index 5d531b3c6..e4670778e 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ConnectSuccessInterceptor.java +++ b/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,16 @@ public boolean exitAfterHandlingConnect(Channel channel, NettyResponseFuture final Future 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, proxyServer, + 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); diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Redirect30xInterceptor.java b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Redirect30xInterceptor.java index df373c32d..7d2a9705f 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Redirect30xInterceptor.java +++ b/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) { diff --git a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java index d205ee739..bdf05133c 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -1313,6 +1313,11 @@ private ListenableFuture 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. diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/ConnectionOrientedAuthHttp2Test.java b/client/src/test/java/org/asynchttpclient/netty/channel/ConnectionOrientedAuthHttp2Test.java new file mode 100644 index 000000000..728475acb --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/channel/ConnectionOrientedAuthHttp2Test.java @@ -0,0 +1,139 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.channel; + +import io.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.Realm; +import org.asynchttpclient.netty.ssl.DefaultSslEngineFactory; +import org.asynchttpclient.uri.Uri; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import javax.net.ssl.SSLEngine; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The registry poll, the registration and the ALPN advertisement all key off the predicate pinned here. + */ +@ExtendWith(NettyLeakDetectorExtension.class) +class ConnectionOrientedAuthHttp2Test { + + @Test + void theConnectionOrientedSchemesAreRecognised() { + for (Realm.AuthScheme scheme : new Realm.AuthScheme[]{ + Realm.AuthScheme.NTLM, Realm.AuthScheme.KERBEROS, Realm.AuthScheme.SPNEGO}) { + assertTrue(PrincipalScopedPartitionKey.authenticatesTheConnection(realm(scheme, "alice")), + scheme + " authenticates the connection, so its connection must not be shared"); + } + } + + /** + * An ambient login carries no principal and the ticket still belongs to someone. + */ + @Test + void aRealmWithNoPrincipalStillAuthenticatesTheConnection() { + assertTrue(PrincipalScopedPartitionKey.authenticatesTheConnection(realm(Realm.AuthScheme.KERBEROS, null))); + assertTrue(PrincipalScopedPartitionKey.authenticatesTheConnection(realm(Realm.AuthScheme.SPNEGO, null))); + } + + @Test + void requestScopedSchemesAndNoRealmAreUnaffected() { + assertFalse(PrincipalScopedPartitionKey.authenticatesTheConnection(null), + "a request with no realm must keep using the HTTP/2 registry"); + for (Realm.AuthScheme scheme : new Realm.AuthScheme[]{ + Realm.AuthScheme.BASIC, Realm.AuthScheme.DIGEST, Realm.AuthScheme.SCRAM_SHA_256}) { + assertFalse(PrincipalScopedPartitionKey.authenticatesTheConnection(realm(scheme, "alice")), + scheme + " authenticates the request, so multiplexing stays available"); + } + } + + /** + * The h2 side ignores the principal; the h1 key cannot. + */ + @Test + void theHttp1PoolStillSeparatesPrincipals() { + Object base = "http://example.com"; + Object alice = PrincipalScopedPartitionKey.scope(base, realm(Realm.AuthScheme.NTLM, "alice")); + Object bob = PrincipalScopedPartitionKey.scope(base, realm(Realm.AuthScheme.NTLM, "bob")); + assertNotEquals(alice, bob, "one principal's HTTP/1.1 connection must not be polled for another"); + assertNotEquals(base, alice); + } + + /** + * A CONNECT tunnel is one socket, and the proxy realm authenticates the tunnel itself. The gates read + * both realms, so a tunnel opened under an NTLM proxy is unshareable even when the origin needs no + * credentials -- the case where the victim request carries no realm of its own. + */ + @Test + void anAuthenticatedProxyHopAlsoMakesTheConnectionUnshareable() { + Realm ntlmProxy = realm(Realm.AuthScheme.NTLM, "alice"); + assertTrue(PrincipalScopedPartitionKey.authenticatesTheConnection(null, ntlmProxy), + "a tunnel the proxy authenticated must not be handed to another principal's request"); + assertTrue(PrincipalScopedPartitionKey.authenticatesTheConnection( + realm(Realm.AuthScheme.BASIC, "bob"), ntlmProxy)); + assertFalse(PrincipalScopedPartitionKey.authenticatesTheConnection( + realm(Realm.AuthScheme.BASIC, "bob"), realm(Realm.AuthScheme.BASIC, "alice")), + "request-scoped schemes on both hops keep multiplexing available"); + assertFalse(PrincipalScopedPartitionKey.authenticatesTheConnection(null, null)); + } + + /** + * Pin that the flag reaches the engine rather than being dropped by the overload. + */ + @Test + void theAlpnDecisionReachesTheSslEngine() { + List asked = new ArrayList<>(); + AsyncHttpClientConfig config = config().setSslEngineFactory(new DefaultSslEngineFactory() { + @Override + public SSLEngine newSslEngine(AsyncHttpClientConfig cfg, String peerHost, int peerPort, boolean http2Allowed) { + asked.add(http2Allowed); + return super.newSslEngine(cfg, peerHost, peerPort, http2Allowed); + } + }).build(); + Timer timer = new HashedWheelTimer(); + ChannelManager channelManager = new ChannelManager(config, timer); + EmbeddedChannel refused = new EmbeddedChannel(); + EmbeddedChannel allowed = new EmbeddedChannel(); + try { + Uri uri = Uri.create("https://example.com:12345/"); + channelManager.addSslHandler(refused.pipeline(), uri, null, false, false); + channelManager.addSslHandler(allowed.pipeline(), uri, null, false, true); + assertEquals(Arrays.asList(false, true), asked, + "a connection-oriented realm must not advertise h2, and everything else still may"); + } finally { + refused.finishAndReleaseAll(); + allowed.finishAndReleaseAll(); + channelManager.close(); + timer.stop(); + } + } + + private static Realm realm(Realm.AuthScheme scheme, String principal) { + return new Realm.Builder(principal, "pass").setScheme(scheme).build(); + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/request/ConnectionOrientedAuthConnectListenerTest.java b/client/src/test/java/org/asynchttpclient/netty/request/ConnectionOrientedAuthConnectListenerTest.java new file mode 100644 index 000000000..e66bc0752 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/request/ConnectionOrientedAuthConnectListenerTest.java @@ -0,0 +1,220 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.request; + +import io.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHandler; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.Realm; +import org.asynchttpclient.Request; +import org.asynchttpclient.RequestBuilder; +import org.asynchttpclient.Response; +import org.asynchttpclient.channel.ChannelPoolPartitioning; +import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.channel.ChannelManager; +import org.asynchttpclient.netty.channel.ConnectionSemaphore; +import org.asynchttpclient.netty.channel.NettyConnectListener; +import org.asynchttpclient.netty.timeout.TimeoutsHolder; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * h2c has no ALPN to refuse h2 for us, so the upgrade decision itself has to carry it. + */ +@ExtendWith(NettyLeakDetectorExtension.class) +class ConnectionOrientedAuthConnectListenerTest { + + private static final String H2C_URL = "http://example.com:12345"; + + private AsyncHttpClientConfig h2cConfig; + private ChannelManager channelManager; + private NettyRequestSender requestSender; + private Timer timer; + + @BeforeEach + void setUp() { + h2cConfig = config().setHttp2Enabled(true).setHttp2CleartextEnabled(true).build(); + timer = new HashedWheelTimer(); + channelManager = new ChannelManager(h2cConfig, timer); + // The listener writes the request once the connection is up, so it needs a real sender; the write + // lands in the EmbeddedChannel and is discarded with it. + requestSender = new NettyRequestSender(h2cConfig, channelManager, timer, null); + } + + @AfterEach + void tearDown() { + if (channelManager != null) { + channelManager.close(); + } + if (timer != null) { + timer.stop(); + } + } + + @Test + void anNtlmRealmKeepsTheCleartextConnectionOffTheHttp2Registry() throws Exception { + CountingSemaphore semaphore = new CountingSemaphore(); + NettyResponseFuture future = newFuture(semaphore, realm(Realm.AuthScheme.NTLM, "alice"), null); + future.acquirePartitionLockLazily(); + + NettyConnectListener listener = new NettyConnectListener<>(future, requestSender, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + + assertNull(channel.pipeline().get(ChannelManager.HTTP2_FRAME_CODEC), + "the connection must stay on HTTP/1.1: upgrading it and then hiding it would strand the " + + "socket and break the NTLM 401 retry, which needs the socket that issued the challenge"); + assertNull(channelManager.pollHttp2Connection(future.getPartitionKey()), + "a socket NTLM authenticates must not be published for another principal to multiplex onto"); + assertEquals(0, semaphore.released.get(), + "the exchange still holds its permit; the point is that it is releasable, not released"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * The proxy hop authenticates the same socket, so a realm-less request through an NTLM proxy is the case + * that matters: without this the connection is registered and any later request draws it. + */ + @Test + void anNtlmProxyRealmAlsoKeepsItOff() throws Exception { + CountingSemaphore semaphore = new CountingSemaphore(); + NettyResponseFuture future = newFuture(semaphore, null, realm(Realm.AuthScheme.NTLM, "alice")); + future.acquirePartitionLockLazily(); + + NettyConnectListener listener = new NettyConnectListener<>(future, requestSender, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + assertNull(channel.pipeline().get(ChannelManager.HTTP2_FRAME_CODEC), + "the proxy hop authenticates the same socket, so this must stay on HTTP/1.1 too: the 407 " + + "retry needs the socket that issued the challenge"); + assertNull(channelManager.pollHttp2Connection(future.getPartitionKey()), + "a tunnel the proxy authenticated must not be shared either"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * The control: without such a realm the connection is still registered, so the guard has not simply + * turned h2c off. + */ + @Test + void withoutSuchARealmTheConnectionIsStillRegistered() throws Exception { + CountingSemaphore semaphore = new CountingSemaphore(); + NettyResponseFuture future = newFuture(semaphore, realm(Realm.AuthScheme.BASIC, "bob"), null); + future.acquirePartitionLockLazily(); + + NettyConnectListener listener = new NettyConnectListener<>(future, requestSender, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + assertNotNull(channel.pipeline().get(ChannelManager.HTTP2_FRAME_CODEC), + "sanity: without such a realm the connection really is upgraded, so the assertions above " + + "are about the guard rather than about h2c being off"); + assertNotNull(channelManager.pollHttp2Connection(future.getPartitionKey()), + "a request-scoped scheme must keep multiplexing"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * An h2 parent is never offered to the HTTP/1.1 pool, so had the guard left it upgraded-but-unregistered + * the permit would be held until the server dropped the socket. + */ + @Test + void closingTheConnectionReturnsThePermit() throws Exception { + CountingSemaphore semaphore = new CountingSemaphore(); + NettyResponseFuture future = newFuture(semaphore, realm(Realm.AuthScheme.KERBEROS, null), null); + future.acquirePartitionLockLazily(); + + NettyConnectListener listener = new NettyConnectListener<>(future, requestSender, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + channel.finishAndReleaseAll(); + + assertEquals(semaphore.acquired.get(), semaphore.released.get(), + "the permit must come back when the socket goes, or maxConnectionsPerHost wedges the host"); + } + + private NettyResponseFuture newFuture(ConnectionSemaphore semaphore, Realm realm, Realm proxyRealm) { + Request request = new RequestBuilder().setUrl(H2C_URL).build(); + NettyResponseFuture future = new NettyResponseFuture<>(request, noopHandler(), + new NettyRequestFactory(h2cConfig).newNettyRequest(request, false, null, realm, proxyRealm), 0, + ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, semaphore, null); + future.setTimeoutsHolder(new TimeoutsHolder(null, future, null, h2cConfig, null)); + future.setRealm(realm); + future.setProxyRealm(proxyRealm); + return future; + } + + private static AsyncHandler noopHandler() { + return new AsyncCompletionHandler() { + @Override + public Object onCompleted(Response response) { + return null; + } + }; + } + + private static Realm realm(Realm.AuthScheme scheme, String principal) { + return new Realm.Builder(principal, "pass").setScheme(scheme).build(); + } + + /** + * Counts acquire/release rather than reading permits, so a release that never happens and one that + * happens twice are both visible. + */ + private static final class CountingSemaphore implements ConnectionSemaphore { + + private final AtomicInteger acquired = new AtomicInteger(); + private final AtomicInteger released = new AtomicInteger(); + + @Override + public void acquireChannelLock(Object partitionKey) throws IOException { + acquired.incrementAndGet(); + } + + @Override + public void acquireChannelLock(Object partitionKey, boolean nonBlocking) throws IOException { + acquired.incrementAndGet(); + } + + @Override + public void releaseChannelLock(Object partitionKey) { + released.incrementAndGet(); + } + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/request/ConnectionOrientedAuthPollGateTest.java b/client/src/test/java/org/asynchttpclient/netty/request/ConnectionOrientedAuthPollGateTest.java new file mode 100644 index 000000000..8f72dadfc --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/request/ConnectionOrientedAuthPollGateTest.java @@ -0,0 +1,172 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.request; + +import io.netty.channel.Channel; +import io.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHandler; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.Realm; +import org.asynchttpclient.Request; +import org.asynchttpclient.RequestBuilder; +import org.asynchttpclient.Response; +import org.asynchttpclient.channel.ChannelPoolPartitioning; +import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.channel.ChannelManager; +import org.asynchttpclient.proxy.ProxyServer; +import org.jetbrains.annotations.Nullable; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.lang.reflect.Method; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * The registration skip does nothing about the other direction: a realm-less request registers a connection, + * an NTLM request draws it and authenticates a socket anyone can reach, and the next realm-less request rides + * it as that principal. Only the poll gate stops that. + */ +@ExtendWith(NettyLeakDetectorExtension.class) +class ConnectionOrientedAuthPollGateTest { + + private static final String URL = "https://example.com:443/"; + + private AsyncHttpClientConfig config; + private ChannelManager channelManager; + private NettyRequestSender sender; + private Timer timer; + + @BeforeEach + void setUp() { + config = config().build(); + timer = new HashedWheelTimer(); + channelManager = new ChannelManager(config, timer); + sender = new NettyRequestSender(config, channelManager, timer, null); + } + + @AfterEach + void tearDown() { + if (channelManager != null) { + channelManager.close(); + } + if (timer != null) { + timer.stop(); + } + } + + @Test + void aConnectionOrientedRealmDoesNotDrawARegisteredHttp2Connection() throws Exception { + EmbeddedChannel registered = new EmbeddedChannel(); + try { + Request request = new RequestBuilder().setUrl(URL).build(); + Object key = request.getChannelPoolPartitioning() + .getPartitionKey(request.getUri(), request.getVirtualHost(), null); + channelManager.registerHttp2Connection(key, registered); + + assertNull(poll(realm(Realm.AuthScheme.NTLM, "alice")), + "authenticating a connection anyone can reach hands every later request that identity"); + assertNull(poll(realm(Realm.AuthScheme.KERBEROS, null)), + "an ambient Kerberos login binds the socket just the same"); + assertNotNull(poll(realm(Realm.AuthScheme.BASIC, "bob")), + "sanity: a request-scoped scheme still multiplexes, so this pins the gate"); + } finally { + registered.finishAndReleaseAll(); + } + } + + @Test + void aConnectionOrientedProxyRealmDoesNotDrawOneEither() throws Exception { + EmbeddedChannel registered = new EmbeddedChannel(); + try { + Request request = new RequestBuilder().setUrl(URL).build(); + Object key = request.getChannelPoolPartitioning() + .getPartitionKey(request.getUri(), request.getVirtualHost(), null); + channelManager.registerHttp2Connection(key, registered); + + NettyResponseFuture future = newFuture(realm(Realm.AuthScheme.BASIC, "bob"), + realm(Realm.AuthScheme.NTLM, "alice")); + assertNull(pollWith(future), "a request whose proxy realm authenticates the connection must not " + + "multiplex onto a shared one either"); + } finally { + registered.finishAndReleaseAll(); + } + } + + /** + * ROUND_ROBIN pins reuse to one IP through a key override, which is a separate branch with its own copy + * of the gate. + */ + @Test + void theRoundRobinOverridePathIsGatedToo() throws Exception { + EmbeddedChannel registered = new EmbeddedChannel(); + try { + Object override = "round-robin-override-key"; + channelManager.registerHttp2Connection(override, registered); + + NettyResponseFuture ntlm = newFuture(realm(Realm.AuthScheme.NTLM, "alice"), null); + ntlm.setPartitionKeyOverride(override); + assertNull(pollWith(ntlm), "the override branch must apply the same gate as the base-key branch"); + + NettyResponseFuture basic = newFuture(realm(Realm.AuthScheme.BASIC, "bob"), null); + basic.setPartitionKeyOverride(override); + assertNotNull(pollWith(basic), "sanity: the override branch still multiplexes otherwise"); + } finally { + registered.finishAndReleaseAll(); + } + } + + private @Nullable Channel poll(Realm realm) throws Exception { + return pollWith(newFuture(realm, null)); + } + + private @Nullable Channel pollWith(NettyResponseFuture future) throws Exception { + Method pollPooledChannel = NettyRequestSender.class.getDeclaredMethod("pollPooledChannel", + NettyResponseFuture.class, Request.class, ProxyServer.class, AsyncHandler.class); + pollPooledChannel.setAccessible(true); + return (Channel) pollPooledChannel.invoke(sender, future, future.getCurrentRequest(), null, noopHandler()); + } + + private NettyResponseFuture newFuture(Realm realm, Realm proxyRealm) { + Request request = new RequestBuilder().setUrl(URL).build(); + NettyResponseFuture future = new NettyResponseFuture<>(request, noopHandler(), null, 0, + ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + future.setRealm(realm); + future.setProxyRealm(proxyRealm); + return future; + } + + private static AsyncHandler noopHandler() { + return new AsyncCompletionHandler() { + @Override + public Object onCompleted(Response response) { + return null; + } + }; + } + + private static Realm realm(Realm.AuthScheme scheme, String principal) { + return new Realm.Builder(principal, "pass").setScheme(scheme).build(); + } +} From dda419980fbd61ed6af588c24fbe063ab3eea055 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Sun, 20 Sep 2026 20:10:35 +0000 Subject: [PATCH 2/2] Address comments --- .../org/asynchttpclient/netty/channel/ChannelManager.java | 5 ++--- .../netty/handler/intercept/ConnectSuccessInterceptor.java | 3 +-- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index f9c6e8d4b..55354e5b8 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -934,14 +934,13 @@ public Future updatePipelineForHttpTunneling(ChannelPipeline pipeline, } public Future updatePipelineForHttpsTunneling(ChannelPipeline pipeline, Uri requestUri, ProxyServer proxyServer) { - return updatePipelineForHttpsTunneling(pipeline, requestUri, proxyServer, true); + return updatePipelineForHttpsTunneling(pipeline, requestUri, true); } /** * @param http2Allowed false when the realm authenticates the socket the tunnel runs over. */ - public Future updatePipelineForHttpsTunneling(ChannelPipeline pipeline, Uri requestUri, ProxyServer proxyServer, - boolean http2Allowed) { + public Future updatePipelineForHttpsTunneling(ChannelPipeline pipeline, Uri requestUri, boolean http2Allowed) { Future whenHandshaked = null; // Remove HTTP codec as tunnel is established diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ConnectSuccessInterceptor.java b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ConnectSuccessInterceptor.java index e4670778e..9057c95cb 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ConnectSuccessInterceptor.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ConnectSuccessInterceptor.java @@ -62,8 +62,7 @@ public boolean exitAfterHandlingConnect(Channel channel, NettyResponseFuture // 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, - http2Allowed); + whenHandshaked = channelManager.updatePipelineForHttpsTunneling(channel.pipeline(), requestUri, http2Allowed); } else { // Standard HTTP proxy or SOCKS proxy tunneling whenHandshaked = channelManager.updatePipelineForHttpTunneling(channel.pipeline(), requestUri, http2Allowed);