From 18980cb43291c2ad3dc93e1383072f9a6173867c Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Thu, 13 Aug 2026 20:23:28 +0300 Subject: [PATCH 1/3] [fix][proxy] PIP-478: resolve the proxy's broker-client credential off the event loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DirectProxyHandler.channelActive called the v4 authentication plugin inline — getAuthData() then authenticate(INIT_AUTH_DATA) — on the thread delivering the channel-active event, and handleAuthChallenge did the same for every challenge round. That is arbitrary plugin code: an OAuth2 token endpoint round trip, an Athenz ZTS fetch, a GSSAPI exchange with the KDC. While it ran, every connection multiplexed onto that Netty loop stalled — the exact hazard PIP-478 removed from the client, left in place on the proxy because pip-478.md put the proxy's broker-client credential I/O out of scope. Now that the client has no synchronous path at all, keeping one here has no justification, so the proxy drives the same v5 machinery: ProxyService owns one V5BinaryAuthenticationDriver built from its started v4 plugin, and each backend connection opens its own exchange against it. The exchange's calls always off-load; with no ClientAuthenticationServices bound — the proxy is not a PulsarClient and has no executor to lend — the work lands on the framework's shared blocking pool, which is the case V5AuthContexts documents for exactly this caller. Command ordering on the channel is unchanged: the continuation is dispatched back onto the channel's own event loop, so the connect command and auth responses are still built and written there. A resolution failure now closes the backend channel rather than only logging, since the proxy has no credential to send and leaving the connection open would wait out the broker's timeout instead of letting the client retry. The broker-pushed REFRESH sentinel starts a fresh exchange, per binary routing rule 2, mirroring ClientCnx. Covered by the existing proxy authentication suites, which exercise both the connect and the REFRESH paths end to end (ProxyRefreshAuthTest, ProxyForwardAuthDataTest, ProxyAuthenticatedProducerConsumerTest, ProxyWithAuthorizationTest — 19 tests, all passing). A dedicated assertion that no credential call lands on a proxy IO thread is still worth adding. --- .../proxy/server/DirectProxyHandler.java | 118 ++++++++++++------ .../pulsar/proxy/server/ProxyService.java | 28 +++++ 2 files changed, 111 insertions(+), 35 deletions(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java index 19e37cba3a748..b8b9dc90cac28 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java @@ -42,14 +42,15 @@ import io.netty.util.CharsetUtil; import java.net.InetSocketAddress; import java.util.Arrays; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import java.util.function.Function; import lombok.CustomLog; import lombok.Getter; import lombok.SneakyThrows; import org.apache.pulsar.PulsarVersion; import org.apache.pulsar.client.api.Authentication; -import org.apache.pulsar.client.api.AuthenticationDataProvider; -import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange; import org.apache.pulsar.common.allocator.PulsarByteBufAllocator; import org.apache.pulsar.common.api.AuthData; import org.apache.pulsar.common.api.proto.BaseCommand; @@ -61,6 +62,7 @@ import org.apache.pulsar.common.protocol.PulsarDecoder; import org.apache.pulsar.common.stats.Rate; import org.apache.pulsar.common.tls.impl.TlsContextAcquisition; +import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.netty.NettyChannelUtil; @CustomLog @@ -80,7 +82,6 @@ public class DirectProxyHandler { public static final String TLS_HANDLER = "tls"; private final Authentication authentication; - private AuthenticationDataProvider authenticationDataProvider; private final ProxyService service; private final Runnable onHandshakeCompleteAction; final boolean tlsEnabledWithBroker; @@ -259,6 +260,9 @@ public class ProxyBackendHandler extends PulsarDecoder { private final ProxyConfiguration config; private final int protocolVersion; private final FeatureFlags featureFlags; + // PIP-478: the v5 exchange this backend connection authenticates through. Replaced on a broker-pushed + // REFRESH, which starts a fresh exchange. Only ever touched on this channel's event loop. + private AuthenticationExchange authExchange; public ProxyBackendHandler(ProxyConfiguration config, int protocolVersion, String remoteHostName, FeatureFlags featureFlags) { @@ -275,16 +279,58 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception { if (config.isHaProxyProtocolEnabled()) { writeHAProxyMessage(); } - - // Send the Connect command to broker - authenticationDataProvider = authentication.getAuthData(remoteHostName); - AuthData authData = authenticationDataProvider.authenticate(AuthData.INIT_AUTH_DATA); - ByteBuf command = Commands.newConnect( - authentication.getAuthMethodName(), authData, protocolVersion, - proxyConnection.clientVersion, null /* target broker */, - originalPrincipal, clientAuthData, clientAuthMethod, PulsarVersion.getVersion(), featureFlags); - writeAndFlush(command); isTlsOutboundChannel = ProxyConnection.isTlsChannel(inboundChannel); + + // Send the Connect command to broker. PIP-478: the credential is resolved through a v5 exchange + // rather than by calling the v4 plugin here. This method runs on the Netty event loop, and the v4 + // call it used to make is arbitrary plugin code — an OAuth2 token endpoint round trip, an Athenz + // ZTS fetch, a GSSAPI exchange with the KDC — which stalled every connection multiplexed onto + // that loop for its duration. The exchange's calls always off-load. + authExchange = service.getProxyClientAuthenticationDriver().newAuthenticationExchange(remoteHostName); + sendWhenResolved(authExchange.getAuthDataAsync(), + authData -> Commands.newConnect( + authentication.getAuthMethodName(), authData, protocolVersion, + proxyConnection.clientVersion, null /* target broker */, + originalPrincipal, clientAuthData, clientAuthMethod, PulsarVersion.getVersion(), + featureFlags), + "connect"); + } + + /** + * Send a command built from an asynchronously-resolved credential (PIP-478). + * + *

The continuation is dispatched onto this channel's event loop, so the command is built and + * written there whether the credential was already in memory or needed I/O — command ordering on the + * channel is therefore unchanged from the synchronous version. A failure closes the backend channel: + * the proxy has no credential to send, and leaving the connection open would wait out the broker's + * timeout instead of letting the client retry. + * + * @param resolution the credential being resolved + * @param commandBuilder builds the command to send from the resolved credential + * @param what what is being authenticated, for logging + */ + private void sendWhenResolved(CompletableFuture resolution, + Function commandBuilder, String what) { + resolution.whenCompleteAsync((authData, throwable) -> { + if (throwable != null) { + Throwable cause = FutureUtil.unwrapCompletionException(throwable); + log.error().attr("channel", ctx.channel()).attr("stage", what).exception(cause) + .log("Failed to resolve the proxy's broker-client credential"); + ctx.close(); + return; + } + if (!ctx.channel().isActive()) { + // The backend connection went away while the credential was resolving. + return; + } + try { + writeAndFlush(commandBuilder.apply(authData)); + } catch (Throwable t) { + log.error().attr("channel", ctx.channel()).attr("stage", what).exception(t) + .log("Failed to send the proxy's broker-client authentication command"); + ctx.close(); + } + }, ctx.executor()); } @Override @@ -345,37 +391,39 @@ protected void handleAuthChallenge(CommandAuthChallenge authChallenge) { checkArgument(authChallenge.hasChallenge()); checkArgument(authChallenge.getChallenge().hasAuthData() && authChallenge.getChallenge().hasAuthData()); - if (Arrays.equals(AuthData.REFRESH_AUTH_DATA_BYTES, authChallenge.getChallenge().getAuthData())) { - try { - authenticationDataProvider = authentication.getAuthData(remoteHostName); - } catch (PulsarClientException e) { - log.error().attr("channel", ctx.channel()) - .exception(e) - .log("Error refreshing authentication data provider"); - return; + // PIP-478 binary routing rule 2: the broker's REFRESH sentinel restarts authentication with a + // fresh exchange whose getAuthDataAsync() re-produces the current credential, rather than being + // routed into the conversation it just terminated. Any other challenge is a round of the current + // exchange, whose state slot carries conversation state across rounds. This mirrors ClientCnx. + boolean refresh = + Arrays.equals(AuthData.REFRESH_AUTH_DATA_BYTES, authChallenge.getChallenge().getAuthData()); + CompletableFuture resolution; + try { + if (refresh) { + authExchange = + service.getProxyClientAuthenticationDriver().newAuthenticationExchange(remoteHostName); + resolution = authExchange.getAuthDataAsync(); + } else { + resolution = authExchange + .authenticateAsync(AuthData.of(authChallenge.getChallenge().getAuthData())); } + } catch (Throwable t) { + // One try/catch so a plugin that throws synchronously fails the connection rather than + // propagating up the event loop. + resolution = CompletableFuture.failedFuture(t); } // mutual authn. If auth not complete, continue auth; if auth complete, complete connectionFuture. - try { - AuthData authData = authenticationDataProvider - .authenticate(AuthData.of(authChallenge.getChallenge().getAuthData())); - + sendWhenResolved(resolution, authData -> { checkState(!authData.isComplete()); - - ByteBuf request = Commands.newAuthResponse(authentication.getAuthMethodName(), - authData, - this.protocolVersion, - PulsarVersion.getVersion()); - log.debug().attr("channel", ctx.channel()) .attr("authMethod", authentication.getAuthMethodName()) .log("Mutual auth"); - - writeAndFlush(request); - } catch (Exception e) { - log.error().exception(e).log("Error mutual verify"); - } + return Commands.newAuthResponse(authentication.getAuthMethodName(), + authData, + this.protocolVersion, + PulsarVersion.getVersion()); + }, "challenge"); } @Override diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java index 5662d59a303e0..eb0815885042b 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java @@ -69,6 +69,9 @@ import org.apache.pulsar.broker.topiclistlimit.TopicListSizeResultCache; import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlets; import org.apache.pulsar.client.api.Authentication; +import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver; +import org.apache.pulsar.client.impl.auth.v5.V5AuthenticationLoader; +import org.apache.pulsar.client.impl.auth.v5.V5BinaryAuthenticationDriver; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.apache.pulsar.client.impl.tls.ClientTlsFactorySupport; import org.apache.pulsar.common.allocator.PulsarByteBufAllocator; @@ -96,6 +99,8 @@ public class ProxyService implements Closeable { private final ProxyConfiguration proxyConfig; private final Authentication proxyClientAuthentication; + // PIP-478: lazily built from proxyClientAuthentication; see getProxyClientAuthenticationDriver(). + private BinaryAuthenticationDriver proxyClientAuthenticationDriver; @Getter private final DnsAddressResolverGroup dnsAddressResolverGroup; @Getter @@ -679,6 +684,29 @@ public Authentication getProxyClientAuthenticationPlugin() { return this.proxyClientAuthentication; } + /** + * The v5 driver the proxy's broker connections authenticate through (PIP-478). + * + *

The proxy owns one started v4 plugin for all of its broker connections, so it owns one driver: + * every {@code DirectProxyHandler} opens its own exchange against it, and the exchange is what carries + * per-connection conversation state. Built from the started plugin, because {@link ProxyService} + * starts and closes that instance itself — the bridge must not run that lifecycle a second time. + * + *

No {@code ClientAuthenticationServices} are bound: the proxy is not a {@code PulsarClient} and has + * no client-owned executor to lend. Credential work therefore lands on the framework's shared blocking + * pool, which is the case {@code V5AuthContexts} documents for exactly this caller — the alternative, + * running it inline, is the Netty event loop. + * + * @return the shared binary authentication driver + */ + public synchronized BinaryAuthenticationDriver getProxyClientAuthenticationDriver() { + if (proxyClientAuthenticationDriver == null) { + proxyClientAuthenticationDriver = new V5BinaryAuthenticationDriver( + V5AuthenticationLoader.forStartedV4Plugin(proxyClientAuthentication)); + } + return proxyClientAuthenticationDriver; + } + public synchronized PrometheusMetricsServlet getMetricsServlet() { return metricsServlet; } From f9ded8beea616a8234fa7686dd73ab7735e1b750 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Fri, 14 Aug 2026 19:23:08 +0300 Subject: [PATCH 2/3] [fix][proxy] PIP-478: serialize and bound the proxy's broker-client auth rounds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AuthenticationExchange is single-round and non-thread-safe, and its javadoc makes serializing rounds the caller's obligation. DirectProxyHandler became its second caller after ClientCnx and did not honour it: while the frame decoder is still running (state Init), two challenge frames delivered in one read reach handleAuthChallenge in the same event-loop turn, before either resolution has completed, and both drive the same exchange. A challenge arriving while a round is in flight is now dropped, which makes rounds strictly serialized — so this class needs none of the generation guarding ClientCnx carries, because nothing left can supersede an in-flight round. ClientCnx needs it only because it lets a REFRESH supersede one; the proxy can drop a REFRESH instead, both because it cannot reach this handler (the connection stops decoding after the handshake) and because the broker's refresh check is a scheduleAtFixedRate task that re-sends on the next tick. Also bounds the exchange with MAX_AUTH_CHALLENGE_ROUNDS, mirroring ClientCnx: a broker that answers every CommandAuthResponse with another challenge would otherwise loop against the proxy forever, and each round now also schedules credential work onto a blocking pool. getProxyClientAuthenticationDriver() now reads the memoized driver lock-free and synchronizes only on the miss, as ClientCnx.resolveAuthDriver does: every backend channelActive would otherwise take the ProxyService monitor, which is shared with the metrics-servlet accessors. The field is volatile accordingly. Adds DirectProxyHandlerAuthTest, which builds ProxyBackendHandler on an EmbeddedChannel — whose event loop runs a submitted task only when the test asks it to, turning "on the event loop or off it" from a race into an assertion, and letting a challenge be delivered while a resolution is deliberately still pending. All four tests are mutation-verified: inlining either v4 credential call fails its off-load test alone with the event-loop thread recorded; removing the drop guard or the round cap fails its own test alone. Smaller review points: the discarded whenCompleteAsync future and the inactive-channel early return now say what they are, and a comment records why the auth method name is read from the v4 plugin rather than from the exchange. Assisted-by: Claude Code (Opus 5) --- .../proxy/server/DirectProxyHandler.java | 66 +++- .../pulsar/proxy/server/ProxyService.java | 24 +- .../server/DirectProxyHandlerAuthTest.java | 300 ++++++++++++++++++ 3 files changed, 382 insertions(+), 8 deletions(-) create mode 100644 pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/DirectProxyHandlerAuthTest.java diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java index b8b9dc90cac28..ce5d50c7e1e9a 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java @@ -81,6 +81,12 @@ public class DirectProxyHandler { private final String clientAuthMethod; public static final String TLS_HANDLER = "tls"; + // PIP-478: hard cap on broker challenge rounds within a single binary authentication exchange, mirroring + // ClientCnx.MAX_AUTH_CHALLENGE_ROUNDS (both in turn mirror HttpAuthenticationDriver.MAX_CHALLENGE_ROUNDS). + // A broker that answers every CommandAuthResponse with another challenge would otherwise loop against the + // proxy forever, and each round now also schedules credential work onto a blocking pool. + static final int MAX_AUTH_CHALLENGE_ROUNDS = 10; + private final Authentication authentication; private final ProxyService service; private final Runnable onHandshakeCompleteAction; @@ -263,6 +269,15 @@ public class ProxyBackendHandler extends PulsarDecoder { // PIP-478: the v5 exchange this backend connection authenticates through. Replaced on a broker-pushed // REFRESH, which starts a fresh exchange. Only ever touched on this channel's event loop. private AuthenticationExchange authExchange; + // PIP-478: round state. AuthenticationExchange is single-round and non-thread-safe, and serializing + // its rounds is the caller's obligation; this class is its second caller after ClientCnx. While the + // frame decoder is still running (state Init) two challenge frames arriving in one read reach + // handleAuthChallenge in the same event-loop turn, before either resolution has completed, and would + // otherwise drive the same exchange concurrently. Both fields are touched only on this channel's + // event loop (channelActive, handleAuthChallenge, and the continuations dispatched there), so they + // need no synchronization. + private boolean authRoundInProgress; + private int authChallengeRounds; public ProxyBackendHandler(ProxyConfiguration config, int protocolVersion, String remoteHostName, FeatureFlags featureFlags) { @@ -286,6 +301,11 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception { // call it used to make is arbitrary plugin code — an OAuth2 token endpoint round trip, an Athenz // ZTS fetch, a GSSAPI exchange with the KDC — which stalled every connection multiplexed onto // that loop for its duration. The exchange's calls always off-load. + // + // The auth method name is read from the v4 plugin here and in handleAuthChallenge below, where + // ClientCnx instead takes it from the exchange that produced the credential. Both are correct for + // the proxy — it owns one started plugin, and the bridge's authMethodName() delegates straight to + // it — and reading it from the plugin does not depend on a round having completed. Deliberate. authExchange = service.getProxyClientAuthenticationDriver().newAuthenticationExchange(remoteHostName); sendWhenResolved(authExchange.getAuthDataAsync(), authData -> Commands.newConnect( @@ -303,7 +323,12 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception { * written there whether the credential was already in memory or needed I/O — command ordering on the * channel is therefore unchanged from the synchronous version. A failure closes the backend channel: * the proxy has no credential to send, and leaving the connection open would wait out the broker's - * timeout instead of letting the client retry. + * timeout instead of letting the client retry. That covers a failure to build the command too, not + * just a failure to resolve the credential. + * + *

This is where an authentication round begins: the round is marked in progress so that a + * challenge arriving before it completes is dropped rather than re-entering the exchange + * concurrently. * * @param resolution the credential being resolved * @param commandBuilder builds the command to send from the resolved credential @@ -311,7 +336,12 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception { */ private void sendWhenResolved(CompletableFuture resolution, Function commandBuilder, String what) { + authRoundInProgress = true; + // The future returned by whenCompleteAsync is intentionally discarded: the continuation handles + // every outcome itself, and the only way that future fails is ctx.executor() rejecting during + // event-loop shutdown — at which point the channel is already going away. resolution.whenCompleteAsync((authData, throwable) -> { + authRoundInProgress = false; if (throwable != null) { Throwable cause = FutureUtil.unwrapCompletionException(throwable); log.error().attr("channel", ctx.channel()).attr("stage", what).exception(cause) @@ -320,7 +350,10 @@ private void sendWhenResolved(CompletableFuture resolution, return; } if (!ctx.channel().isActive()) { - // The backend connection went away while the credential was resolving. + // The backend connection went away while the credential was resolving. Logged so that + // "backend connected but never sent CommandConnect" is diagnosable rather than silent. + log.debug().attr("channel", ctx.channel()).attr("stage", what) + .log("Backend channel closed while the proxy's broker-client credential resolved"); return; } try { @@ -395,8 +428,37 @@ protected void handleAuthChallenge(CommandAuthChallenge authChallenge) { // fresh exchange whose getAuthDataAsync() re-produces the current credential, rather than being // routed into the conversation it just terminated. Any other challenge is a round of the current // exchange, whose state slot carries conversation state across rounds. This mirrors ClientCnx. + // The REFRESH branch is conformance with that rule rather than a path the broker can reach here: + // this handler stops decoding once state == HandshakeCompleted, and the broker arms its refresh + // task with an initial delay of authenticationRefreshCheckSeconds after connect completes, so by + // the time a REFRESH is pushed it is proxied straight through to the client, which answers it. + // The proxy's own credential refresh lives in ProxyConnection, not here. boolean refresh = Arrays.equals(AuthData.REFRESH_AUTH_DATA_BYTES, authChallenge.getChallenge().getAuthData()); + // PIP-478 (serialize-or-drop): a broker never pipelines challenges — it waits for each + // CommandAuthResponse — so a challenge arriving while a round is still in flight is anomalous, + // and servicing it would re-enter the same single-round, non-thread-safe exchange concurrently. + // Dropping it makes rounds strictly serialized, which is why this class needs none of the + // generation guarding ClientCnx carries: nothing here can supersede an in-flight round. ClientCnx + // does need it, because it lets a REFRESH supersede one; the proxy can drop a REFRESH instead, + // both because it cannot reach this handler and because the broker's refresh check is a + // scheduleAtFixedRate task that would re-send it on the next tick. + if (authRoundInProgress) { + log.debug().attr("channel", ctx.channel()) + .log("Dropping a broker auth challenge received while an auth round is in progress"); + return; + } + // Bound the exchange. A REFRESH opens a fresh exchange, so it resets the counter; any other + // challenge counts towards the cap. + if (refresh) { + authChallengeRounds = 0; + } else if (++authChallengeRounds > MAX_AUTH_CHALLENGE_ROUNDS) { + log.error().attr("channel", ctx.channel()).attr("maxChallengeRounds", MAX_AUTH_CHALLENGE_ROUNDS) + .log("Binary authentication exceeded the maximum challenge rounds; closing the " + + "broker connection"); + ctx.close(); + return; + } CompletableFuture resolution; try { if (refresh) { diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java index eb0815885042b..fb781c35df605 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java @@ -100,7 +100,8 @@ public class ProxyService implements Closeable { private final ProxyConfiguration proxyConfig; private final Authentication proxyClientAuthentication; // PIP-478: lazily built from proxyClientAuthentication; see getProxyClientAuthenticationDriver(). - private BinaryAuthenticationDriver proxyClientAuthenticationDriver; + // Volatile because that getter reads it without holding the monitor. + private volatile BinaryAuthenticationDriver proxyClientAuthenticationDriver; @Getter private final DnsAddressResolverGroup dnsAddressResolverGroup; @Getter @@ -697,14 +698,25 @@ public Authentication getProxyClientAuthenticationPlugin() { * pool, which is the case {@code V5AuthContexts} documents for exactly this caller — the alternative, * running it inline, is the Netty event loop. * + *

Read on every backend connection's {@code channelActive}, so the hit path is lock-free and only a + * miss takes the monitor — as {@code ClientCnx.resolveAuthDriver} does, and for the same reason: this + * monitor is the {@link ProxyService} one, shared with the metrics-servlet accessors, and a connection + * being set up should not have to queue behind unrelated machinery. + * * @return the shared binary authentication driver */ - public synchronized BinaryAuthenticationDriver getProxyClientAuthenticationDriver() { - if (proxyClientAuthenticationDriver == null) { - proxyClientAuthenticationDriver = new V5BinaryAuthenticationDriver( - V5AuthenticationLoader.forStartedV4Plugin(proxyClientAuthentication)); + public BinaryAuthenticationDriver getProxyClientAuthenticationDriver() { + BinaryAuthenticationDriver resolved = proxyClientAuthenticationDriver; + if (resolved != null) { + return resolved; + } + synchronized (this) { + if (proxyClientAuthenticationDriver == null) { + proxyClientAuthenticationDriver = new V5BinaryAuthenticationDriver( + V5AuthenticationLoader.forStartedV4Plugin(proxyClientAuthentication)); + } + return proxyClientAuthenticationDriver; } - return proxyClientAuthenticationDriver; } public synchronized PrometheusMetricsServlet getMetricsServlet() { diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/DirectProxyHandlerAuthTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/DirectProxyHandlerAuthTest.java new file mode 100644 index 0000000000000..e1d9b9f39fb37 --- /dev/null +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/DirectProxyHandlerAuthTest.java @@ -0,0 +1,300 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.pulsar.proxy.server; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.embedded.EmbeddedChannel; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; +import org.apache.pulsar.client.api.Authentication; +import org.apache.pulsar.client.api.AuthenticationDataProvider; +import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver; +import org.apache.pulsar.client.impl.auth.v5.V5AuthenticationLoader; +import org.apache.pulsar.client.impl.auth.v5.V5BinaryAuthenticationDriver; +import org.apache.pulsar.common.api.AuthData; +import org.apache.pulsar.common.api.proto.CommandAuthChallenge; +import org.apache.pulsar.common.api.proto.FeatureFlags; +import org.apache.pulsar.common.protocol.Commands; +import org.awaitility.Awaitility; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * PIP-478: the proxy's broker-client credential must not be resolved on the Netty event loop, and the + * exchange it is resolved through must have its rounds serialized and bounded. + * + *

Both properties live entirely inside {@code DirectProxyHandler.ProxyBackendHandler}, so they are pinned + * here against an {@link EmbeddedChannel} rather than through a running proxy: an embedded event loop only + * runs the tasks submitted to it when {@code runPendingTasks()} is called, which makes "did this run on the + * event loop or off it" an assertion rather than a race. An end-to-end proxy fixture can observe neither — + * the handshake completes the same way whichever thread the plugin was called on. + */ +public class DirectProxyHandlerAuthTest { + + private static final String AUTH_METHOD = "thread-recording"; + private static final String BROKER_HOST = "broker.example:6650"; + + private ProxyConfiguration proxyConfig; + private ProxyService service; + private ProxyConnection proxyConnection; + private EmbeddedChannel inboundChannel; + private EmbeddedChannel backendChannel; + private ThreadRecordingAuthentication plugin; + + /** + * A v4 plugin that records the thread each of its credential calls runs on, standing in for one that + * blocks there — an OAuth2 token endpoint round trip, an Athenz ZTS fetch, a GSSAPI exchange with the KDC. + * It answers challenges too, so a challenge round reaches {@code authenticate} a second time. + */ + private static final class ThreadRecordingAuthentication implements Authentication { + + private final List credentialThreads = new CopyOnWriteArrayList<>(); + + @Override + public String getAuthMethodName() { + return AUTH_METHOD; + } + + @Override + public AuthenticationDataProvider getAuthData(String brokerHostName) { + return new AuthenticationDataProvider() { + @Override + public boolean hasDataFromCommand() { + return true; + } + + @Override + public AuthData authenticate(AuthData data) { + credentialThreads.add(Thread.currentThread().getName()); + return AuthData.of("credential".getBytes(UTF_8)); + } + }; + } + + @SuppressWarnings("deprecation") + @Override + public void configure(Map authParams) { + } + + @Override + public void start() { + } + + @Override + public void close() { + } + } + + /** + * A driver whose rounds the test completes by hand, for the guards that are about when a round is + * serviced rather than about which thread services it. + */ + private static final class ScriptedDriver implements BinaryAuthenticationDriver { + + private final Supplier> connectCredential; + private final Supplier> challengeResponse; + private final AtomicInteger challengeRounds = new AtomicInteger(); + + private ScriptedDriver(Supplier> connectCredential, + Supplier> challengeResponse) { + this.connectCredential = connectCredential; + this.challengeResponse = challengeResponse; + } + + @Override + public AuthenticationExchange newAuthenticationExchange(String brokerHostName) { + return new AuthenticationExchange() { + @Override + public CompletableFuture getAuthDataAsync() { + return connectCredential.get(); + } + + @Override + public String authMethodName() { + return AUTH_METHOD; + } + + @Override + public CompletableFuture authenticateAsync(AuthData challenge) { + challengeRounds.incrementAndGet(); + return challengeResponse.get(); + } + }; + } + } + + @BeforeMethod + public void setUp() { + proxyConfig = new ProxyConfiguration(); + plugin = new ThreadRecordingAuthentication(); + inboundChannel = new EmbeddedChannel(); + + ChannelHandlerContext inboundCtx = mock(ChannelHandlerContext.class); + when(inboundCtx.channel()).thenReturn(inboundChannel); + + service = mock(ProxyService.class); + when(service.getConfiguration()).thenReturn(proxyConfig); + + proxyConnection = mock(ProxyConnection.class); + when(proxyConnection.ctx()).thenReturn(inboundCtx); + when(proxyConnection.getClientAuthentication()).thenReturn(plugin); + proxyConnection.clientVersion = "test-client"; + } + + @AfterMethod(alwaysRun = true) + public void tearDown() { + if (backendChannel != null) { + backendChannel.finishAndReleaseAll(); + } + inboundChannel.finishAndReleaseAll(); + } + + /** + * Build the backend handler on an embedded channel and activate it, which is what starts the connect + * round. The outbound channel is published before registration so the handler has somewhere to write. + * + * @param driver the authentication driver the handler resolves its credential through + * @return the backend handler under test + * @throws Exception if activating the channel fails + */ + private DirectProxyHandler.ProxyBackendHandler activateBackend(BinaryAuthenticationDriver driver) + throws Exception { + when(service.getProxyClientAuthenticationDriver()).thenReturn(driver); + DirectProxyHandler directProxyHandler = new DirectProxyHandler(service, proxyConnection); + DirectProxyHandler.ProxyBackendHandler backend = directProxyHandler.new ProxyBackendHandler( + proxyConfig, Commands.getCurrentProtocolVersion(), BROKER_HOST, new FeatureFlags()); + backendChannel = new EmbeddedChannel(false, false, backend); + directProxyHandler.outboundChannel = backendChannel; + backendChannel.register(); + return backend; + } + + private BinaryAuthenticationDriver realDriver() { + return new V5BinaryAuthenticationDriver(V5AuthenticationLoader.forStartedV4Plugin(plugin)); + } + + private static CommandAuthChallenge challenge(String payload) { + CommandAuthChallenge command = new CommandAuthChallenge(); + command.setProtocolVersion(Commands.getCurrentProtocolVersion()) + .setChallenge() + .setAuthData(payload.getBytes(UTF_8)) + .setAuthMethodName(AUTH_METHOD); + return command; + } + + /** Drain the embedded loop until the handler has written the command it owes, or fail. */ + private void awaitCommandWritten(int expectedCommands) { + Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> { + backendChannel.runPendingTasks(); + assertThat(backendChannel.outboundMessages()).hasSize(expectedCommands); + }); + } + + @Test + public void theConnectCredentialIsResolvedOffTheEventLoop() throws Exception { + activateBackend(realDriver()); + awaitCommandWritten(1); + + assertThat(plugin.credentialThreads) + .as("the v4 credential for CommandConnect must not be resolved on the channel's event loop") + .singleElement().asString().startsWith("pulsar-auth-blocking-shared"); + } + + /** + * The second half of the composition. Off-loading only the connect credential would leave every challenge + * round — a SASL exchange with the KDC, say — running on the loop that carries all of this proxy's + * multiplexed connections. + */ + @Test + public void theChallengeRoundIsResolvedOffTheEventLoop() throws Exception { + DirectProxyHandler.ProxyBackendHandler backend = activateBackend(realDriver()); + awaitCommandWritten(1); + + backend.handleAuthChallenge(challenge("server-round-1")); + awaitCommandWritten(2); + + assertThat(plugin.credentialThreads) + .as("both v4 credential calls must be resolved off the channel's event loop") + .hasSize(2) + .allSatisfy(thread -> assertThat(thread).startsWith("pulsar-auth-blocking-shared")); + } + + /** + * {@code AuthenticationExchange} is single-round and non-thread-safe, and serializing its rounds is the + * caller's obligation. While the frame decoder is still running, two challenge frames arriving in one read + * reach {@code handleAuthChallenge} in the same event-loop turn — the sequence reproduced here — and + * without the guard both would drive the same exchange before either resolution completed. + */ + @Test + public void aChallengeArrivingWhileARoundIsInFlightIsDropped() throws Exception { + CompletableFuture connectCredential = new CompletableFuture<>(); + ScriptedDriver driver = new ScriptedDriver(() -> connectCredential, + () -> CompletableFuture.completedFuture(AuthData.of("response".getBytes(UTF_8)))); + DirectProxyHandler.ProxyBackendHandler backend = activateBackend(driver); + + // The connect round is still resolving: nothing has completed it yet. + backend.handleAuthChallenge(challenge("early-challenge")); + backendChannel.runPendingTasks(); + assertThat(driver.challengeRounds).as("a challenge received mid-round must not re-enter the exchange") + .hasValue(0); + + // Once the round in flight lands, the next challenge is serviced normally. + connectCredential.complete(AuthData.of("credential".getBytes(UTF_8))); + awaitCommandWritten(1); + backend.handleAuthChallenge(challenge("later-challenge")); + awaitCommandWritten(2); + assertThat(driver.challengeRounds).as("a challenge received between rounds must be serviced") + .hasValue(1); + } + + /** + * A broker that answers every {@code CommandAuthResponse} with another challenge would otherwise loop + * against the proxy forever, and each round now also schedules credential work onto a blocking pool. + */ + @Test + public void anEndlesslyChallengingBrokerIsCutOffAtTheRoundCap() throws Exception { + ScriptedDriver driver = new ScriptedDriver( + () -> CompletableFuture.completedFuture(AuthData.of("credential".getBytes(UTF_8))), + () -> CompletableFuture.completedFuture(AuthData.of("response".getBytes(UTF_8)))); + DirectProxyHandler.ProxyBackendHandler backend = activateBackend(driver); + awaitCommandWritten(1); + + for (int round = 1; round <= DirectProxyHandler.MAX_AUTH_CHALLENGE_ROUNDS + 1; round++) { + backend.handleAuthChallenge(challenge("server-round-" + round)); + backendChannel.runPendingTasks(); + } + + assertThat(driver.challengeRounds) + .as("the exchange must stop at the round cap rather than answering challenges forever") + .hasValue(DirectProxyHandler.MAX_AUTH_CHALLENGE_ROUNDS); + assertThat(backendChannel.isOpen()) + .as("exceeding the round cap must close the backend connection") + .isFalse(); + } +} From 3186c4acef0ef4bdf6877df9c3dae3c49f2d3f1d Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Thu, 20 Aug 2026 03:19:53 +0300 Subject: [PATCH 3/3] [improve][pip] PIP-478: move the proxy data path out of Out of Scope Out of Scope still said DirectProxyHandler "resolves the credential inline on the proxy's Netty loop, on connect and on the broker's refresh sentinel", and that closing it "requires an async rework of that inline getAuthData()". This PR is that rework, so the scope contract the PIP was voted on now understates what shipped: an operator reading it would believe a proxy configured with OAuth2 or Athenz still stalls its event loop on every backend connect and every refresh sentinel, and might keep sizing around a hazard that is gone. It also left the serialization and round-cap invariants this PR adds to DirectProxyHandler with no design text behind them. The bullet now describes what the data path does -- a per-connection exchange on the ProxyService-owned V5BinaryAuthenticationDriver, the continuation dispatched back to the channel's event loop, serialize-or-drop challenge rounds, the shared round cap -- and keeps only the three items still genuinely deferred: neither proxy leg binds the framework HTTP client factory, both use the shared library-owned pool rather than a client-owned bounded executor, and the proxy startup ordering is unchanged. Retitled accordingly, since the credential I/O itself is no longer the gap. Assisted-by: Claude Code (Opus 5) --- pip/pip-478.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pip/pip-478.md b/pip/pip-478.md index 1ecc5ea72cc64..1ec0a7e9effff 100644 --- a/pip/pip-478.md +++ b/pip/pip-478.md @@ -142,7 +142,7 @@ This PIP introduces new API and implementation across the existing `pulsar-clien - **Non-Java client SDKs.** Each non-Java SDK (Python, Go, C++, Node.js) follows its own auth model and will be addressed by per-SDK PIPs. -- **Off-loading the proxy's own broker-client credential I/O (known gap).** Motivation #1's event-loop-safety guarantee is delivered for `PulsarClient` / `PulsarAdmin` (and the broker's outbound clients, which are genuine clients that bind the framework's `ClientAuthenticationServices`). The **proxy's** connection to the broker is not a `PulsarClientImpl`: its lookup path uses a bare `ConnectionPool` and its data path a hand-rolled Netty `DirectProxyHandler`, and neither binds the client auth services (bounded blocking executor, framework HTTP client factory). On the proxy's **lookup** path the credential no longer runs on that loop: `ClientCnx` resolves and memoizes a binary authentication driver for a configuration that no `PulsarClient` owns, and with no services bound the credential call falls back to the framework's process-wide shared blocking pool. What remains is the **data** path: `DirectProxyHandler` resolves the credential inline on the proxy's Netty loop, on connect and on the broker's refresh sentinel — **the same behavior as v4, not a regression introduced by this PIP**. Closing it requires an async rework of that inline `getAuthData()`; the lookup leg additionally still runs without the framework HTTP client factory and on a shared library-owned pool rather than a client-owned bounded executor, which is why reordering proxy startup (the broker-client `Authentication` is created and started before the proxy's event loop, DNS resolver, and TLS factory exist) is still part of the follow-up. Both are deferred rather than bundled into this change. +- **Binding the proxy's broker-client legs to client-owned services (known gap).** Motivation #1's event-loop-safety guarantee is delivered for `PulsarClient` / `PulsarAdmin` (and the broker's outbound clients, which are genuine clients that bind the framework's `ClientAuthenticationServices`). The **proxy's** connection to the broker is not a `PulsarClientImpl`: its lookup path uses a bare `ConnectionPool` and its data path a hand-rolled Netty `DirectProxyHandler`, and neither binds the client auth services (bounded blocking executor, framework HTTP client factory). On the proxy's **lookup** path the credential no longer runs on that loop: `ClientCnx` resolves and memoizes a binary authentication driver for a configuration that no `PulsarClient` owns, and with no services bound the credential call falls back to the framework's process-wide shared blocking pool. The **data** path is off-loaded too: `DirectProxyHandler` opens a per-connection exchange on a `ProxyService`-owned `V5BinaryAuthenticationDriver` and resolves the credential through it — on connect and on the broker's refresh sentinel alike — dispatching the continuation back to the channel's event loop, with challenge rounds strictly serialized (a non-refresh challenge arriving while a round is in flight is dropped) and bounded by the same round cap `ClientCnx` uses. What remains deferred: neither proxy leg binds the framework HTTP client factory, both run credential work on the shared library-owned pool rather than a client-owned bounded executor, and reordering proxy startup (the broker-client `Authentication` is created and started before the proxy's event loop, DNS resolver, and TLS factory exist) is still follow-up work. - **The broader FIPS-compliance mode.** This PIP covers the **TLS-transport** requirements for FIPS: a configurable TLS engine (JDK, not native BoringSSL) wired through every component plus the two configurable provider axes (Motivation #4, Goal #5). It does **not** define a full FIPS-mode profile: FIPS-approved algorithms in message encryption (key-wrap) and authentication (password hashing, token signing), a FIPS distribution/packaging variant (shipping `bc-fips` and excluding non-validated `bcprov` / `netty-tcnative-boringssl` / Conscrypt), and a fail-fast `fipsMode` validation switch are a **separate effort** — Pulsar-wide in scope and independent of this SPI. Concretely, the shipped `pulsar-server` distribution today bundles the **non-FIPS** BouncyCastle provider (`bcprov-jdk18on` / `bcpkix-jdk18on`) and explicitly excludes `bc-fips`, and it ships no `bctls-fips` (the jar registering `BCJSSE`) at all; because the two BouncyCastle families declare the same `org.bouncycastle.*` classes under different signers they cannot coexist on one classpath, so until that packaging effort lands a FIPS deployment assembles the provider classpath itself. The in-tree `pulsar-client-test-bcfips` module assembles a classpath that way — excluding the non-FIPS BouncyCastle jars in favour of `bc-fips` — but it covers the crypto side only: it ships no `bctls-fips` and sets neither provider key, so it is not an end-to-end FIPS TLS test. This PIP deliberately provides only the TLS-transport configurability those deployments require, so the two efforts compose without one blocking the other.