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. 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..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 @@ -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 @@ -79,8 +81,13 @@ 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 AuthenticationDataProvider authenticationDataProvider; private final ProxyService service; private final Runnable onHandshakeCompleteAction; final boolean tlsEnabledWithBroker; @@ -259,6 +266,18 @@ 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; + // 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) { @@ -275,16 +294,76 @@ 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. + // + // 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( + 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. 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 + * @param what what is being authenticated, for logging + */ + 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) + .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. 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 { + 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 +424,68 @@ 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. + // 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) { + 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..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 @@ -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,9 @@ public class ProxyService implements Closeable { private final ProxyConfiguration proxyConfig; private final Authentication proxyClientAuthentication; + // PIP-478: lazily built from proxyClientAuthentication; see getProxyClientAuthenticationDriver(). + // Volatile because that getter reads it without holding the monitor. + private volatile BinaryAuthenticationDriver proxyClientAuthenticationDriver; @Getter private final DnsAddressResolverGroup dnsAddressResolverGroup; @Getter @@ -679,6 +685,40 @@ 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. + * + *

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 BinaryAuthenticationDriver getProxyClientAuthenticationDriver() { + BinaryAuthenticationDriver resolved = proxyClientAuthenticationDriver; + if (resolved != null) { + return resolved; + } + synchronized (this) { + if (proxyClientAuthenticationDriver == null) { + proxyClientAuthenticationDriver = new V5BinaryAuthenticationDriver( + V5AuthenticationLoader.forStartedV4Plugin(proxyClientAuthentication)); + } + return proxyClientAuthenticationDriver; + } + } + public synchronized PrometheusMetricsServlet getMetricsServlet() { return metricsServlet; } 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(); + } +}