You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/08/13 00:54:22 UTC
(pulsar) 01/03: [improve][client] PIP-478: Athenz and SASL v5 authentication plugins
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-pip-478-auth-plugins-v3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a288fcc241fa9fc5d329b89ecb1a3b92907029f5
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Thu Aug 13 01:27:53 2026 +0300
[improve][client] PIP-478: Athenz and SASL v5 authentication plugins
Migrate the two remaining built-in authentication plugins onto the v5 SPI, so every
built-in now hands the client a v5-native body rather than being bridged.
SASL is the interesting one: it is multi-round on both transports, and SaslAuthenticationV5
is the first production implementor of the framework HTTP authentication driver the core
migration added — the piece that makes HttpAuthenticationDriver / AsyncHttpAuthenticationProvider
live rather than an unused extension point.
- AthenzAuthenticationV5: single-pass role token over both transports. The ZTS exchange and
its cache stay on the v4 shim, which owns the Athenz SDK's transport; the body reads the
current role token through a provider.
- SaslAuthenticationV5: BinaryAuthDataProvider + BinaryAuthChallengeHandler for the binary
protocol, HttpAuthChallengeHandler + HttpAuthHeadersProvider for SASL over HTTP. The
per-broker PulsarSaslClient lives in the exchange's call-context state slot, so one body
serves the whole client while each connection keeps its own handshake state.
- Both shims expose their body through V5AuthenticationProvider, matching token, basic and
OAuth2. The seam's javadoc no longer has to except them.
Two defects found reviewing the original version of this change are folded in:
- JaxRsChallengeTransport leaked every successfully-completed JAX-RS Response.
InvocationCallback hands the caller an unclosed response, and reading only its headers
neither consumes the entity nor releases the connection, so the success branch leaked one
pooled connection per authentication round while only the timed-out branch closed. The
driver runs at least one round on every admin request, so this was per request, not per
client. Completion and close now both go through completeAndClose().
- AuthenticationSasl.client and saslRoleToken were plain fields, written by start()/close()
on the application thread and read from the challenge driver's Jersey continuation threads
with no happens-before edge. Both are volatile, matching the neighbouring fields.
A third fix from that review is deliberately dropped: it made PulsarClientBuilderV5 drive a
plugin implementing AsyncAuthenticationDriver raw rather than wrapping it in
V5ToV4AuthenticationAdapter, to stop wrapping from hiding the plugin's HTTP capabilities.
The v5-native inversion removed both the wrapper and the decision — the builder now always
hands the raw plugin to the v4 slot and the client derives the body — so the branch and its
tests no longer describe anything the code does.
Assisted-by: Claude Code (Opus 5)
---
.../api/v5/internal/V5AuthenticationProvider.java | 7 +-
.../client/impl/auth/AuthenticationAthenz.java | 62 ++++-
.../impl/auth/v5/AthenzAuthenticationV5.java | 117 ++++++++
.../pulsar/client/impl/auth/v5/package-info.java | 21 +-
.../client/impl/auth/AuthenticationAthenzTest.java | 42 +++
pulsar-client-auth-sasl/build.gradle.kts | 3 +
.../client/impl/auth/AuthenticationSasl.java | 165 ++++++++++-
.../client/impl/auth/v5/SaslAuthenticationV5.java | 308 +++++++++++++++++++++
.../pulsar/client/impl/auth/v5/package-info.java} | 22 +-
.../impl/auth/v5/SaslAuthenticationV5HttpTest.java | 305 ++++++++++++++++++++
10 files changed, 1016 insertions(+), 36 deletions(-)
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/internal/V5AuthenticationProvider.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/internal/V5AuthenticationProvider.java
index c6ddd38925b9..13a12d98bb45 100644
--- a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/internal/V5AuthenticationProvider.java
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/internal/V5AuthenticationProvider.java
@@ -26,10 +26,9 @@ import org.apache.pulsar.client.api.v5.auth.Authentication;
*
* <p>The client drives the v5 authentication model natively: it resolves exactly one v5
* {@link Authentication} per client and drives every transport from it. A v4 plugin therefore has to be
- * turned into a v5 one, and there are two ways to do that. The built-in shims that already <em>have</em> a
- * v5-native body — token, basic and OAuth2, where the v4 class is a thin compatibility surface over it —
- * hand that body over through this interface. Everything else, including the built-in Athenz and SASL
- * plugins, whose credential machinery has no v5-native body today, is wrapped by
+ * turned into a v5 one, and there are two ways to do that. The built-in shims already <em>have</em> a
+ * v5-native body — the v4 class is a thin compatibility surface over it — so they hand that body over
+ * through this interface. Everything else is wrapped by
* {@code LegacyV4AuthenticationAdapter}, which re-expresses the v4 surface as v5 capabilities and
* off-loads every v4 call to the client's blocking executor.
*
diff --git a/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenz.java b/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenz.java
index f4b1bc135a1b..aba6c347157c 100644
--- a/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenz.java
+++ b/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenz.java
@@ -41,6 +41,7 @@ import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.PrivateKey;
import java.util.Map;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
@@ -52,14 +53,28 @@ import org.apache.pulsar.client.api.EncodedAuthenticationParameterSupport;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientException.GettingAuthenticationDataException;
import org.apache.pulsar.client.api.url.URL;
+import org.apache.pulsar.client.api.v5.internal.V5AuthenticationProvider;
import org.apache.pulsar.client.impl.AuthenticationUtil;
+import org.apache.pulsar.client.impl.auth.v5.AthenzAuthenticationV5;
-public class AuthenticationAthenz implements Authentication, EncodedAuthenticationParameterSupport {
+/**
+ * Athenz authentication provider.
+ *
+ * <p>The verbatim v4 synchronous surface ({@link #getAuthData()} / {@link AuthenticationDataAthenz}) is
+ * preserved for callers of the v4 API; PIP-478 additionally exposes the v5-native body through
+ * {@link V5AuthenticationProvider}, which is what the client drives. The old note: it can drive
+ * the role-token credential over the non-blocking binary path via the v5-native
+ * {@link AthenzAuthenticationV5}. The ZTS role-token cache stays on this shim.
+ */
+public class AuthenticationAthenz
+ implements Authentication, EncodedAuthenticationParameterSupport, V5AuthenticationProvider {
private static final long serialVersionUID = 1L;
private static final String APPLICATION_X_PEM_FILE = "application/x-pem-file";
+ // PIP-478: the client's framework services, late-bound before start(); null until then.
+
private transient KeyRefresher keyRefresher = null;
private transient ZTSClient ztsClient = null;
private String ztsUrl = null;
@@ -135,6 +150,51 @@ public class AuthenticationAthenz implements Authentication, EncodedAuthenticati
return (System.nanoTime() - cachedRoleTokenTimestamp) < TimeUnit.MINUTES.toNanos(cacheDurationInMinutes);
}
+ @Override
+ public org.apache.pulsar.client.api.v5.auth.Authentication v5Authentication() {
+ // PIP-478: the client drives this v5-native body. The ZTS role-token exchange and its cache stay on
+ // this shim — it runs on the Athenz SDK's own transport, which the framework does not own — and the
+ // body reads the current role token through the provider below.
+ return new AthenzAuthenticationV5(new ShimRoleTokenProvider(this));
+ }
+
+ @SuppressWarnings("deprecation")
+ private String currentRoleToken() {
+ try {
+ return getAuthData().getCommandData();
+ } catch (PulsarClientException e) {
+ // Wrap in CompletionException, not a bare RuntimeException: on the async binary path this runs
+ // inside V5AuthContexts.supplyBlocking, and BinaryAuthenticationExchange strips exactly one
+ // CompletionException layer before mapping to the v4 exception type. A RuntimeException would
+ // survive that single unwrap and flatten a GettingAuthenticationDataException (transient
+ // credential-acquisition failure) into a generic PulsarClientException.
+ throw new CompletionException(e);
+ }
+ }
+
+ private String currentRoleHeaderName() {
+ return isNotBlank(roleHeader) ? roleHeader : ZTSClient.getHeader();
+ }
+
+ private static final class ShimRoleTokenProvider implements AthenzAuthenticationV5.RoleTokenProvider {
+ private static final long serialVersionUID = 1L;
+ private final AuthenticationAthenz shim;
+
+ ShimRoleTokenProvider(AuthenticationAthenz shim) {
+ this.shim = shim;
+ }
+
+ @Override
+ public String roleToken() {
+ return shim.currentRoleToken();
+ }
+
+ @Override
+ public String roleHeaderName() {
+ return shim.currentRoleHeaderName();
+ }
+ }
+
@Override
public void configure(String encodedAuthParamString) {
checkArgument(isNotBlank(encodedAuthParamString), "authParams must not be empty");
diff --git a/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/v5/AthenzAuthenticationV5.java b/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/v5/AthenzAuthenticationV5.java
new file mode 100644
index 000000000000..4d7d3d0f2015
--- /dev/null
+++ b/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/v5/AthenzAuthenticationV5.java
@@ -0,0 +1,117 @@
+/*
+ * 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.client.impl.auth.v5;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import org.apache.pulsar.client.api.v5.auth.AuthenticationCallContext;
+import org.apache.pulsar.client.api.v5.auth.AuthenticationInitContext;
+import org.apache.pulsar.client.api.v5.auth.BinaryAuthData;
+import org.apache.pulsar.client.api.v5.auth.HttpAuthCallContext;
+import org.apache.pulsar.client.api.v5.auth.HttpAuthHeaders;
+import org.apache.pulsar.client.api.v5.auth.SinglePassAuthentication;
+
+/**
+ * v5-native Athenz authentication (PIP-478). A single-pass credential served over both transports: the
+ * binary protocol sends the current ZTS role token as {@code auth_data}, and HTTP sends the role token
+ * under the configured Athenz role header. The built-in v4 {@code AuthenticationAthenz} drives this body
+ * on the async binary path; the ZTS role-token acquisition and the 90-minute role-token cache remain on
+ * the v4 shim, which supplies the current role token and the HTTP role header to this body via the
+ * role-token provider.
+ *
+ * <p>Role-token acquisition performs network I/O. As with the existing v4 Athenz code, the provider may
+ * block on the ZTS client; off-loading the blocking call belongs to the framework layer, so
+ * this body introduces no threading of its own.
+ *
+ * <p>Although the v5 {@code Authentication} SPI deliberately does not extend {@link Serializable}, this
+ * concrete built-in body is serializable so the v4 {@code AuthenticationAthenz} shim (whose interface
+ * requires {@code Serializable}, for Functions/connector frameworks) round-trips. The role-token provider
+ * is itself serializable.
+ */
+public class AthenzAuthenticationV5 implements SinglePassAuthentication, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The stable auth-method name. */
+ public static final String AUTH_METHOD_NAME = "athenz";
+
+ private final RoleTokenProvider roleTokenProvider;
+
+ // Late-bound at initializeAsync(...): the client's bounded blocking executor, onto which the
+ // (ZTS-blocking) role-token fetch is off-loaded so it never runs on the Netty event loop
+ // (PIP-478). Null when used outside a client, in which case the fetch runs inline.
+ private transient volatile Executor blockingExecutor;
+
+ /**
+ * @param roleTokenProvider supplies the current role token and its HTTP header name on each call
+ */
+ public AthenzAuthenticationV5(RoleTokenProvider roleTokenProvider) {
+ this.roleTokenProvider = roleTokenProvider;
+ }
+
+ @Override
+ public String authMethodName() {
+ return AUTH_METHOD_NAME;
+ }
+
+ @Override
+ public CompletableFuture<Void> initializeAsync(AuthenticationInitContext ctx) {
+ this.blockingExecutor = ctx.blockingExecutor();
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<BinaryAuthData> getAuthDataAsync(AuthenticationCallContext ctx) {
+ return V5AuthContexts.supplyBlocking(blockingExecutor,
+ () -> new BinaryAuthData(roleTokenProvider.roleToken().getBytes(StandardCharsets.UTF_8)));
+ }
+
+ @Override
+ public CompletableFuture<HttpAuthHeaders> getHttpHeadersAsync(HttpAuthCallContext ctx) {
+ return V5AuthContexts.supplyBlocking(blockingExecutor, () -> {
+ Map<String, String> headers = new LinkedHashMap<>();
+ headers.put(roleTokenProvider.roleHeaderName(), roleTokenProvider.roleToken());
+ return HttpAuthHeaders.of(headers);
+ });
+ }
+
+ @Override
+ public void close() {
+ }
+
+ /**
+ * Supplies the current Athenz role token (acquiring/refreshing as needed) and the HTTP header name
+ * under which it is sent. Implementations must be serializable.
+ */
+ public interface RoleTokenProvider extends Serializable {
+ /**
+ * @return the current ZTS role token
+ */
+ String roleToken();
+
+ /**
+ * @return the HTTP header name carrying the role token
+ */
+ String roleHeaderName();
+ }
+}
diff --git a/pulsar-client-auth-sasl/build.gradle.kts b/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/v5/package-info.java
similarity index 65%
copy from pulsar-client-auth-sasl/build.gradle.kts
copy to pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/v5/package-info.java
index 894f9ba66a33..4b9b443afc75 100644
--- a/pulsar-client-auth-sasl/build.gradle.kts
+++ b/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/v5/package-info.java
@@ -16,17 +16,10 @@
* specific language governing permissions and limitations
* under the License.
*/
-
-plugins {
- id("pulsar.public-java-library-conventions")
-}
-
-dependencies {
- implementation(libs.slog)
- compileOnly(project(":pulsar-client-original"))
- implementation(project(":pulsar-common"))
- implementation(libs.guava)
- implementation(libs.commons.lang3)
- implementation(libs.jakarta.ws.rs.api)
- implementation(libs.jersey.client)
-}
+/**
+ * v5-native implementation of the built-in Athenz authentication plugin (PIP-478). The v4
+ * {@code AuthenticationAthenz} class in the parent package is a thin shim that keeps its verbatim
+ * synchronous surface and drives {@link org.apache.pulsar.client.impl.auth.v5.AthenzAuthenticationV5} on
+ * the async binary path via the shared {@code V5BinaryAuthenticationDriver}.
+ */
+package org.apache.pulsar.client.impl.auth.v5;
diff --git a/pulsar-client-auth-athenz/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenzTest.java b/pulsar-client-auth-athenz/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenzTest.java
index cc514fd2f047..91abafd599d5 100644
--- a/pulsar-client-auth-athenz/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenzTest.java
+++ b/pulsar-client-auth-athenz/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenzTest.java
@@ -45,7 +45,14 @@ import java.util.Base64;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import lombok.Cleanup;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
+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.util.ObjectMapperFactory;
import org.mockito.MockedConstruction;
import org.mockito.Mockito;
@@ -346,4 +353,39 @@ public class AuthenticationAthenzTest {
auth3.close();
}
}
+
+ @Test
+ public void testAsyncPathPreservesGettingAuthenticationDataException() throws Exception {
+ // PIP-478: a ZTS failure surfaces from getAuthData() as GettingAuthenticationDataException. On the
+ // async binary path (the v5 body the client drives -> getAuthDataAsync), the exchange strips one
+ // CompletionException layer before mapping back to the v4 exception type, so currentRoleToken() must
+ // re-wrap the v4 exception in a CompletionException; otherwise the transient credential-acquisition
+ // subtype would be flattened to a generic PulsarClientException.
+ final String paramsStr = new String(Files.readAllBytes(Paths.get("./src/test/resources/authParams.json")));
+ try (MockedConstruction<ZTSClient> mockedZTSClient = Mockito.mockConstruction(ZTSClient.class,
+ (mock, context) -> when(mock.getRoleToken(any(), any(), anyInt(), anyInt(), anyBoolean()))
+ .thenThrow(new RuntimeException("ZTS unavailable")))) {
+ final AuthenticationAthenz auth = new AuthenticationAthenz();
+ auth.configure(paramsStr);
+
+ // Resolve the body exactly as the client does, then drive it through the same driver ClientCnx
+ // uses, so this exercises the production path rather than a test-only seam.
+ final AuthenticationExchange exchange =
+ new V5BinaryAuthenticationDriver(V5AuthenticationLoader.forStartedV4Plugin(auth))
+ .newAuthenticationExchange("broker.example.com");
+ // No blocking executor is bound in this test, so supplyBlocking runs the credential fetch inline
+ // and the failure is produced deterministically here.
+ final CompletableFuture<AuthData> future = exchange.getAuthDataAsync();
+ assertTrue(future.isCompletedExceptionally());
+ try {
+ future.join();
+ fail("expected the ZTS failure to propagate");
+ } catch (CompletionException ce) {
+ assertTrue(ce.getCause() instanceof PulsarClientException.GettingAuthenticationDataException,
+ "expected a v4 GettingAuthenticationDataException subtype, got: " + ce.getCause());
+ }
+
+ auth.close();
+ }
+ }
}
diff --git a/pulsar-client-auth-sasl/build.gradle.kts b/pulsar-client-auth-sasl/build.gradle.kts
index 894f9ba66a33..f495a0a6654b 100644
--- a/pulsar-client-auth-sasl/build.gradle.kts
+++ b/pulsar-client-auth-sasl/build.gradle.kts
@@ -29,4 +29,7 @@ dependencies {
implementation(libs.commons.lang3)
implementation(libs.jakarta.ws.rs.api)
implementation(libs.jersey.client)
+ // PIP-478 stage 3d: the SASL-over-HTTP conversation test drives the real SaslAuthenticationV5 body
+ // through the framework HttpAuthenticationDriver (both live in pulsar-client-original).
+ testImplementation(project(":pulsar-client-original"))
}
diff --git a/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationSasl.java b/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationSasl.java
index 4c7fe9f2009f..47e6096d2f0d 100644
--- a/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationSasl.java
+++ b/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationSasl.java
@@ -45,12 +45,18 @@ import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import java.io.IOException;
import java.net.HttpURLConnection;
+import java.net.URI;
+import java.time.Duration;
import java.util.Base64;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import javax.security.auth.login.LoginException;
import lombok.CustomLog;
import lombok.SneakyThrows;
@@ -58,8 +64,16 @@ import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.apache.pulsar.client.api.EncodedAuthenticationParameterSupport;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.v5.auth.HttpAuthHeaders;
+import org.apache.pulsar.client.api.v5.internal.ClientAuthenticationServices;
+import org.apache.pulsar.client.api.v5.internal.ClientAuthenticationServicesAware;
+import org.apache.pulsar.client.api.v5.internal.V5AuthenticationProvider;
import org.apache.pulsar.client.impl.AuthenticationUtil;
import org.apache.pulsar.client.impl.auth.PulsarSaslClient.ClientCallbackHandler;
+import org.apache.pulsar.client.impl.auth.v5.AsyncHttpAuthenticationProvider;
+import org.apache.pulsar.client.impl.auth.v5.HttpAuthenticationDriver;
+import org.apache.pulsar.client.impl.auth.v5.HttpChallengeTransport;
+import org.apache.pulsar.client.impl.auth.v5.SaslAuthenticationV5;
import org.apache.pulsar.common.api.AuthData;
import org.apache.pulsar.common.sasl.JAASCredentialsContainer;
@@ -71,7 +85,9 @@ import org.apache.pulsar.common.sasl.JAASCredentialsContainer;
* for Kerberos a krb5.conf, which is set by `-Djava.security.krb5.conf=/dir/krb5.conf`
*/
@CustomLog
-public class AuthenticationSasl implements Authentication, EncodedAuthenticationParameterSupport {
+public class AuthenticationSasl
+ implements Authentication, EncodedAuthenticationParameterSupport, V5AuthenticationProvider,
+ AsyncHttpAuthenticationProvider, ClientAuthenticationServicesAware {
private static final long serialVersionUID = 1L;
// this is a static object that shares amongst client.
private static JAASCredentialsContainer jaasCredentialsContainer;
@@ -80,6 +96,12 @@ public class AuthenticationSasl implements Authentication, EncodedAuthentication
private Map<String, String> configuration;
private String loginContextName;
private String serverType = null;
+ // PIP-478: the client's framework services, late-bound before start(); null until then.
+ private transient volatile ClientAuthenticationServices authServices;
+ // PIP-478: the framework HTTP auth driver for the SASL-over-HTTP flow, created lazily on the
+ // first HTTP request after start(). Its default transport is this plugin's own JAX-RS client (the
+ // faithful admin-path transport; the lookup path supplies its shared AsyncHttpClient instead).
+ private transient volatile HttpAuthenticationDriver httpAuthenticationDriver;
public AuthenticationSasl() {
}
@@ -102,6 +124,138 @@ public class AuthenticationSasl implements Authentication, EncodedAuthentication
}
}
+ @Override
+ public void bindClientAuthenticationServices(ClientAuthenticationServices services) {
+ this.authServices = services;
+ }
+
+ @Override
+ public org.apache.pulsar.client.api.v5.auth.Authentication v5Authentication() {
+ // PIP-478: the client drives this v5-native body over the binary transport. One body serves the
+ // client; the per-broker PulsarSaslClient that carries a multi-round handshake lives in the
+ // exchange's call-context state slot, not here. The SASL-over-HTTP loop stays on this shim, driven
+ // through httpAuthenticationDriver() below.
+ return new SaslAuthenticationV5(new ShimSaslProviderFactory(this));
+ }
+
+ @Override
+ public Optional<HttpAuthenticationDriver> httpAuthenticationDriver() {
+ // PIP-478: expose the v5-native SASL-over-HTTP body to the framework HTTP auth driver so the
+ // HTTP callers route the 401->resubmit->200 exchange through the shared state machine instead of the
+ // deprecated authenticationStage(...) hook (which stays for third-party v4 plugins). The default
+ // transport is this plugin's own JAX-RS client — exactly what authenticationStage(...) uses today.
+ HttpAuthenticationDriver driver = httpAuthenticationDriver;
+ if (driver == null) {
+ synchronized (this) {
+ driver = httpAuthenticationDriver;
+ if (driver == null) {
+ driver = new HttpAuthenticationDriver(
+ new SaslAuthenticationV5(new ShimSaslProviderFactory(this)),
+ authServices, new JaxRsChallengeTransport());
+ httpAuthenticationDriver = driver;
+ }
+ }
+ }
+ return Optional.of(driver);
+ }
+
+ /**
+ * A {@link HttpChallengeTransport} backed by this plugin's own JAX-RS {@link Client} (created in
+ * {@link #start()}) — the faithful transport for the admin-path SASL warmup, matching what
+ * {@code authenticationStage(...)} does today. Each round is a bodiless {@code GET} to the original URI.
+ * The JAX-RS client itself is created with default, unbounded timeouts, so the per-request {@code timeout}
+ * is enforced here by bounding the returned future ({@link CompletableFuture#orTimeout}) AND cancelling the
+ * underlying JAX-RS {@link Future} on timeout/failure — the request is bounded, so a peer that accepts the
+ * connection but never responds cannot leak an in-flight request (fd/socket exhaustion) across retries. A
+ * response that arrives after the future already timed out is closed rather than leaked.
+ */
+ /**
+ * Complete {@code future} from {@code response} and always close the response.
+ *
+ * <p>{@code InvocationCallback<Response>} hands the caller an unclosed response, and reading only its
+ * headers neither consumes the entity nor releases the connection — so completing without closing leaks
+ * one pooled connection per authentication round. The driver runs at least one round on every admin
+ * request (round 0 replays the cached role token), so the leak is per request, not per client. The late
+ * case — the future already timed out or was cancelled — must close too, which is the only case the
+ * original code handled.
+ *
+ * <p>Package-private (VisibleForTesting) so the close contract can be asserted on both branches.
+ *
+ * @param future the future to complete
+ * @param response the JAX-RS response, always closed before returning
+ */
+
+ private static HttpAuthHeaders toHeaders(Response response) {
+ Map<String, String> headers = new LinkedHashMap<>();
+ response.getStringHeaders().forEach((name, values) -> {
+ if (values != null && !values.isEmpty() && values.get(0) != null) {
+ headers.put(name, values.get(0));
+ }
+ });
+ return HttpAuthHeaders.of(headers);
+ }
+
+ static void completeAndClose(CompletableFuture<HttpChallengeTransport.Result> future, Response response) {
+ try {
+ future.complete(new HttpChallengeTransport.Result(response.getStatus(), toHeaders(response)));
+ } finally {
+ response.close();
+ }
+ }
+
+ private final class JaxRsChallengeTransport implements HttpChallengeTransport {
+ @Override
+ public CompletableFuture<Result> get(URI uri, HttpAuthHeaders requestHeaders, Duration timeout) {
+ CompletableFuture<Result> future = new CompletableFuture<>();
+ try {
+ Client c = client;
+ if (c == null) {
+ throw new IllegalStateException("SASL authentication HTTP client is not started");
+ }
+ Builder builder = c.target(uri).request(MediaType.APPLICATION_JSON);
+ requestHeaders.asMap().forEach(builder::header);
+ Future<Response> responseFuture = builder.async().get(new InvocationCallback<Response>() {
+ @Override
+ public void completed(Response response) {
+ completeAndClose(future, response);
+ }
+
+ @Override
+ public void failed(Throwable throwable) {
+ future.completeExceptionally(throwable);
+ }
+ });
+ if (timeout != null && !timeout.isNegative() && !timeout.isZero()) {
+ future.orTimeout(timeout.toNanos(), TimeUnit.NANOSECONDS);
+ }
+ // Cancel the unbounded JAX-RS request on timeout/failure so its socket is released, not leaked.
+ future.whenComplete((result, throwable) -> {
+ if (!responseFuture.isDone()) {
+ responseFuture.cancel(true);
+ }
+ });
+ } catch (Throwable t) {
+ future.completeExceptionally(t);
+ }
+ return future;
+ }
+
+ }
+
+ private static final class ShimSaslProviderFactory implements SaslAuthenticationV5.SaslProviderFactory {
+ private static final long serialVersionUID = 1L;
+ private final AuthenticationSasl shim;
+
+ ShimSaslProviderFactory(AuthenticationSasl shim) {
+ this.shim = shim;
+ }
+
+ @Override
+ public AuthenticationDataProvider create(String brokerHost) throws Exception {
+ return shim.getAuthData(brokerHost);
+ }
+ }
+
@Override
public void configure(String encodedAuthParamString) {
if (isBlank(encodedAuthParamString)) {
@@ -167,6 +321,8 @@ public class AuthenticationSasl implements Authentication, EncodedAuthentication
client.close();
client = null;
}
+ // Drop the cached HTTP driver; its default transport captured the now-closed JAX-RS client.
+ httpAuthenticationDriver = null;
if (jaasCredentialsContainer != null) {
jaasCredentialsContainer.close();
jaasCredentialsContainer = null;
@@ -174,8 +330,11 @@ public class AuthenticationSasl implements Authentication, EncodedAuthentication
}
}
- private String saslRoleToken = null;
- private Client client = null;
+ // PIP-478: written by start()/close() on the application thread and read from the HTTP challenge
+ // driver's continuation threads (Jersey async callbacks), so both need a happens-before edge —
+ // matching the volatile treatment already given to authServices / httpAuthenticationDriver.
+ private volatile String saslRoleToken = null;
+ private volatile Client client = null;
// role token exists but expired return true
private boolean isRoleTokenExpired(Map<String, String> responseHeaders) {
diff --git a/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/v5/SaslAuthenticationV5.java b/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/v5/SaslAuthenticationV5.java
new file mode 100644
index 000000000000..7e60886baa24
--- /dev/null
+++ b/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/v5/SaslAuthenticationV5.java
@@ -0,0 +1,308 @@
+/*
+ * 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.client.impl.auth.v5;
+
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_AUTH_ROLE_TOKEN;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_AUTH_ROLE_TOKEN_EXPIRED;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_AUTH_TOKEN;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_HEADER_STATE;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_HEADER_TYPE;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_CLIENT_INIT;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_COMPLETE;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_NEGOTIATE;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_SERVER;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_SERVER_CHECK_TOKEN;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_TYPE_VALUE;
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.Executor;
+import org.apache.pulsar.client.api.AuthenticationDataProvider;
+import org.apache.pulsar.client.api.v5.PulsarClientException;
+import org.apache.pulsar.client.api.v5.auth.AuthChallenge;
+import org.apache.pulsar.client.api.v5.auth.Authentication;
+import org.apache.pulsar.client.api.v5.auth.AuthenticationCallContext;
+import org.apache.pulsar.client.api.v5.auth.AuthenticationInitContext;
+import org.apache.pulsar.client.api.v5.auth.BinaryAuthChallengeHandler;
+import org.apache.pulsar.client.api.v5.auth.BinaryAuthData;
+import org.apache.pulsar.client.api.v5.auth.BinaryAuthDataProvider;
+import org.apache.pulsar.client.api.v5.auth.ChallengeResponse;
+import org.apache.pulsar.client.api.v5.auth.HttpAuthCallContext;
+import org.apache.pulsar.client.api.v5.auth.HttpAuthChallengeHandler;
+import org.apache.pulsar.client.api.v5.auth.HttpAuthHeaders;
+import org.apache.pulsar.client.api.v5.auth.HttpAuthHeadersProvider;
+import org.apache.pulsar.common.api.AuthData;
+import org.apache.pulsar.common.sasl.SaslConstants;
+
+/**
+ * v5-native SASL authentication for the Pulsar binary protocol (PIP-478). SASL is a multi-round
+ * challenge/response handshake, so this body implements {@link BinaryAuthDataProvider} for the initial
+ * {@code CommandConnect} credential and {@link BinaryAuthChallengeHandler} for each subsequent
+ * {@code CommandAuthChallenge}. The per-exchange SASL conversation state (an
+ * {@link AuthenticationDataProvider} wrapping a {@code PulsarSaslClient}) is created on the initial call
+ * and kept in the call-context state slot so each broker handshake stays isolated.
+ *
+ * <p>The same body also serves the SASL-over-HTTP flow: it implements {@link HttpAuthChallengeHandler}
+ * (the SASL-style {@code 401}-carrying-custom-headers multi-round exchange) and {@link HttpAuthHeadersProvider}
+ * (the validated role-token replay onto the real request). The framework HTTP auth driver runs the
+ * {@code 401}→resubmit→{@code 200} loop; this body only computes each round's headers, porting the v4
+ * {@code AuthenticationSasl.newRequestHeader} / {@code getHeaders} state machine. Its per-exchange
+ * conversation (the {@code PulsarSaslClient} wrapper plus the captured role token) lives in the HTTP
+ * call-context state slot, so concurrent HTTP handshakes stay isolated. A fresh per-exchange SASL provider
+ * is obtained through the serializable {@link SaslProviderFactory} supplied by the shim (it reads the
+ * shim's JAAS subject and server type).
+ *
+ * <p>Although the v5 {@code Authentication} SPI deliberately does not extend {@link Serializable}, this
+ * concrete built-in body is serializable so the v4 {@code AuthenticationSasl} shim (whose interface
+ * requires {@code Serializable}, for Functions/connector frameworks) round-trips. The provider factory is
+ * itself serializable.
+ */
+public class SaslAuthenticationV5 implements Authentication, BinaryAuthDataProvider,
+ BinaryAuthChallengeHandler, HttpAuthChallengeHandler, HttpAuthHeadersProvider, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The stable auth-method name. */
+ public static final String AUTH_METHOD_NAME = SaslConstants.AUTH_METHOD_NAME;
+
+ private final SaslProviderFactory providerFactory;
+
+ // PIP-478 FIX D: the client's bounded blocking executor, late-bound at initializeAsync(...). The SASL
+ // provider creation and evaluateChallenge/authenticate (GSSAPI/Kerberos) work is off-loaded onto it so
+ // it never runs on the Netty event loop. Null when used outside a client -> degraded inline computation.
+ private transient volatile Executor blockingExecutor;
+
+ // PIP-478 FIX C: the cross-request SASL-over-HTTP role-token cache, restoring the v4
+ // AuthenticationSasl.saslRoleToken semantics. The framework HTTP auth driver reuses ONE body instance
+ // across requests but gives each request a fresh call-context/conversation, so the validated role token
+ // must live on the body to be replayed (State=ServerCheckToken) instead of restarting a full Kerberos
+ // negotiation every request. Cleared and re-negotiated on SaslAuthRoleTokenExpired.
+ private transient volatile String cachedRoleToken;
+
+ /**
+ * @param providerFactory creates a fresh per-exchange SASL data provider for a broker host
+ */
+ public SaslAuthenticationV5(SaslProviderFactory providerFactory) {
+ this.providerFactory = providerFactory;
+ }
+
+ @Override
+ public String authMethodName() {
+ return AUTH_METHOD_NAME;
+ }
+
+ @Override
+ public CompletableFuture<Void> initializeAsync(AuthenticationInitContext ctx) {
+ this.blockingExecutor = ctx.blockingExecutor();
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<BinaryAuthData> getAuthDataAsync(AuthenticationCallContext ctx) {
+ // FIX D: run the SASL provider creation + initial evaluateChallenge off the event loop.
+ return V5AuthContexts.supplyBlocking(blockingExecutor, () -> {
+ try {
+ AuthenticationDataProvider provider = providerFactory.create(ctx.brokerHost());
+ ctx.setStateObject(AuthenticationDataProvider.class, provider);
+ AuthData initData = provider.authenticate(AuthData.INIT_AUTH_DATA);
+ return new BinaryAuthData(initData.getBytes());
+ } catch (Exception e) {
+ throw new CompletionException(e);
+ }
+ });
+ }
+
+ @Override
+ public CompletableFuture<ChallengeResponse> respondToChallengeAsync(AuthenticationCallContext ctx,
+ AuthChallenge authChallenge) {
+ // FIX D: run evaluateChallenge (GSSAPI) off the event loop.
+ return V5AuthContexts.supplyBlocking(blockingExecutor, () -> {
+ try {
+ AuthenticationDataProvider provider =
+ ctx.getStateObject(AuthenticationDataProvider.class).orElse(null);
+ if (provider == null) {
+ // No prior state (e.g. a broker-pushed challenge without a preceding connect call on this
+ // context); start a fresh SASL exchange.
+ provider = providerFactory.create(ctx.brokerHost());
+ ctx.setStateObject(AuthenticationDataProvider.class, provider);
+ }
+ AuthData response = provider.authenticate(AuthData.of(authChallenge.bytes()));
+ return new ChallengeResponse(response.getBytes());
+ } catch (Exception e) {
+ throw new CompletionException(e);
+ }
+ });
+ }
+
+ // ---- SASL-over-HTTP: HttpAuthChallengeHandler + HttpAuthHeadersProvider (PIP-478) ----
+ //
+ // Ports AuthenticationSasl.newRequestHeader / getHeaders. The framework driver runs the
+ // 401->resubmit->200 loop (bodiless GET to the original URI); respondToHttpChallengeAsync computes each
+ // round's request headers from the server's prior response, and getHttpHeadersAsync produces the real
+ // request's headers (the validated role token). Cross-round state (the PulsarSaslClient conversation
+ // plus the role token) lives in the call-context state slot, keyed by SaslHttpConversation.
+
+ /**
+ * The per-exchange SASL-over-HTTP conversation held in the HTTP call-context state slot: the
+ * {@link AuthenticationDataProvider} wrapping this exchange's {@code PulsarSaslClient}, and the role
+ * token once the server has issued it.
+ */
+ private static final class SaslHttpConversation {
+ private AuthenticationDataProvider provider;
+ private String roleToken;
+ }
+
+ @Override
+ public CompletableFuture<HttpAuthHeaders> respondToHttpChallengeAsync(HttpAuthCallContext ctx) {
+ // FIX D: run the SASL negotiation (evaluateChallenge/GSSAPI) off the event loop.
+ return V5AuthContexts.supplyBlocking(blockingExecutor, () -> {
+ try {
+ SaslHttpConversation conv = conversation(ctx);
+ // The server carries its prior response's SASL headers here (empty on the first round).
+ boolean hasPrevious = ctx.serverChallengeHeaders().isPresent();
+ // Capture the role token if the server has issued it (the terminal 200 also carries it).
+ String issuedRoleToken = header(ctx, SASL_AUTH_ROLE_TOKEN);
+ if (issuedRoleToken != null) {
+ conv.roleToken = issuedRoleToken;
+ }
+
+ Map<String, String> headers = new LinkedHashMap<>();
+ // The SASL data provider's HTTP headers (SASL-Type: Kerberos), on every request.
+ conv.provider.getHttpHeaders().forEach(e -> headers.put(e.getKey(), e.getValue()));
+
+ // Role token exists but the server reported it expired: drop it (including the cross-request
+ // cache, FIX C) and restart the SASL exchange.
+ if (isRoleTokenExpired(conv, ctx)) {
+ hasPrevious = false;
+ conv.roleToken = null;
+ cachedRoleToken = null;
+ conv.provider = providerFactory.create(host(ctx));
+ }
+
+ // Role token in hand: replay it, asking the server to check / negotiate / complete.
+ if (conv.roleToken != null) {
+ headers.put(SASL_AUTH_ROLE_TOKEN, conv.roleToken);
+ if (!hasPrevious) {
+ headers.put(SASL_HEADER_STATE, SASL_STATE_SERVER_CHECK_TOKEN);
+ } else if (SASL_STATE_COMPLETE.equalsIgnoreCase(header(ctx, SASL_HEADER_STATE))) {
+ headers.put(SASL_HEADER_STATE, SASL_STATE_COMPLETE);
+ } else {
+ headers.put(SASL_HEADER_STATE, SASL_STATE_NEGOTIATE);
+ }
+ return HttpAuthHeaders.of(headers);
+ }
+
+ // No role token yet: run the SASL negotiation.
+ if (!hasPrevious) {
+ headers.put(SASL_HEADER_STATE, SASL_STATE_CLIENT_INIT);
+ AuthData initData = conv.provider.authenticate(AuthData.INIT_AUTH_DATA);
+ headers.put(SASL_AUTH_TOKEN, Base64.getEncoder().encodeToString(initData.getBytes()));
+ } else {
+ AuthData brokerData = AuthData.of(Base64.getDecoder().decode(header(ctx, SASL_AUTH_TOKEN)));
+ AuthData clientData = conv.provider.authenticate(brokerData);
+ headers.put(SASL_STATE_SERVER, header(ctx, SASL_STATE_SERVER));
+ headers.put(SASL_HEADER_TYPE, SASL_TYPE_VALUE);
+ headers.put(SASL_HEADER_STATE, SASL_STATE_NEGOTIATE);
+ headers.put(SASL_AUTH_TOKEN, Base64.getEncoder().encodeToString(clientData.getBytes()));
+ }
+ return HttpAuthHeaders.of(headers);
+ } catch (Exception e) {
+ throw new CompletionException(e);
+ }
+ });
+ }
+
+ @Override
+ public CompletableFuture<HttpAuthHeaders> getHttpHeadersAsync(HttpAuthCallContext ctx) {
+ try {
+ SaslHttpConversation conv = ctx.getStateObject(SaslHttpConversation.class).orElse(null);
+ // The role token arrives on the terminal 200 response; fall back to any captured during rounds.
+ String roleToken = header(ctx, SASL_AUTH_ROLE_TOKEN);
+ if (roleToken == null && conv != null) {
+ roleToken = conv.roleToken;
+ }
+ if (roleToken == null) {
+ return CompletableFuture.failedFuture(new PulsarClientException.AuthenticationException(
+ "SASL over HTTP exchange completed without issuing a role token"));
+ }
+ if (conv != null) {
+ conv.roleToken = roleToken;
+ }
+ // FIX C: publish the validated role token to the cross-request cache so the next request replays
+ // it (State=ServerCheckToken) instead of restarting a full Kerberos negotiation (v4 semantics).
+ cachedRoleToken = roleToken;
+ Map<String, String> headers = new LinkedHashMap<>();
+ headers.put(SASL_HEADER_TYPE, SASL_TYPE_VALUE);
+ headers.put(SASL_AUTH_ROLE_TOKEN, roleToken);
+ headers.put(SASL_HEADER_STATE, SASL_STATE_COMPLETE);
+ return CompletableFuture.completedFuture(HttpAuthHeaders.of(headers));
+ } catch (Throwable t) {
+ return CompletableFuture.failedFuture(t);
+ }
+ }
+
+ private SaslHttpConversation conversation(HttpAuthCallContext ctx) throws Exception {
+ SaslHttpConversation conv = ctx.getStateObject(SaslHttpConversation.class).orElse(null);
+ if (conv == null) {
+ conv = new SaslHttpConversation();
+ conv.provider = providerFactory.create(host(ctx));
+ // FIX C: seed this request's conversation from the cross-request role-token cache so a
+ // previously validated token is replayed (State=ServerCheckToken) rather than renegotiated.
+ conv.roleToken = cachedRoleToken;
+ ctx.setStateObject(SaslHttpConversation.class, conv);
+ }
+ return conv;
+ }
+
+ // Mirrors AuthenticationSasl.isRoleTokenExpired: a role token exists and the server's prior response
+ // is a Kerberos SASL response whose State reports the role token expired.
+ private static boolean isRoleTokenExpired(SaslHttpConversation conv, HttpAuthCallContext ctx) {
+ return conv.roleToken != null
+ && SASL_TYPE_VALUE.equalsIgnoreCase(header(ctx, SASL_HEADER_TYPE))
+ && SASL_AUTH_ROLE_TOKEN_EXPIRED.equalsIgnoreCase(header(ctx, SASL_HEADER_STATE));
+ }
+
+ private static String header(HttpAuthCallContext ctx, String name) {
+ return ctx.serverChallengeHeaders().flatMap(h -> h.get(name)).orElse(null);
+ }
+
+ private static String host(HttpAuthCallContext ctx) {
+ return ctx.requestUri() == null ? null : ctx.requestUri().getHost();
+ }
+
+ @Override
+ public void close() {
+ }
+
+ /**
+ * Creates a fresh per-exchange SASL {@link AuthenticationDataProvider} (wrapping a new
+ * {@code PulsarSaslClient}) for a given broker host. Implementations must be serializable.
+ */
+ public interface SaslProviderFactory extends Serializable {
+ /**
+ * @param brokerHost the broker host the SASL client authenticates against
+ * @return a fresh SASL data provider for this exchange
+ * @throws Exception if the SASL client could not be created
+ */
+ AuthenticationDataProvider create(String brokerHost) throws Exception;
+ }
+}
diff --git a/pulsar-client-auth-sasl/build.gradle.kts b/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/v5/package-info.java
similarity index 63%
copy from pulsar-client-auth-sasl/build.gradle.kts
copy to pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/v5/package-info.java
index 894f9ba66a33..cc6f3ed5ea3d 100644
--- a/pulsar-client-auth-sasl/build.gradle.kts
+++ b/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/v5/package-info.java
@@ -16,17 +16,11 @@
* specific language governing permissions and limitations
* under the License.
*/
-
-plugins {
- id("pulsar.public-java-library-conventions")
-}
-
-dependencies {
- implementation(libs.slog)
- compileOnly(project(":pulsar-client-original"))
- implementation(project(":pulsar-common"))
- implementation(libs.guava)
- implementation(libs.commons.lang3)
- implementation(libs.jakarta.ws.rs.api)
- implementation(libs.jersey.client)
-}
+/**
+ * v5-native binary-protocol implementation of the built-in SASL authentication plugin (PIP-478). The v4
+ * {@code AuthenticationSasl} class in the parent package is a shim that keeps its verbatim synchronous
+ * surface (including the SASL-over-HTTP loop) and drives
+ * {@link org.apache.pulsar.client.impl.auth.v5.SaslAuthenticationV5} on the async binary path via the
+ * shared {@code V5BinaryAuthenticationDriver}.
+ */
+package org.apache.pulsar.client.impl.auth.v5;
diff --git a/pulsar-client-auth-sasl/src/test/java/org/apache/pulsar/client/impl/auth/v5/SaslAuthenticationV5HttpTest.java b/pulsar-client-auth-sasl/src/test/java/org/apache/pulsar/client/impl/auth/v5/SaslAuthenticationV5HttpTest.java
new file mode 100644
index 000000000000..894c5dd94fd2
--- /dev/null
+++ b/pulsar-client-auth-sasl/src/test/java/org/apache/pulsar/client/impl/auth/v5/SaslAuthenticationV5HttpTest.java
@@ -0,0 +1,305 @@
+/*
+ * 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.client.impl.auth.v5;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_AUTH_ROLE_TOKEN;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_AUTH_ROLE_TOKEN_EXPIRED;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_AUTH_TOKEN;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_HEADER_STATE;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_HEADER_TYPE;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_CLIENT_INIT;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_COMPLETE;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_NEGOTIATE;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_SERVER;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_SERVER_CHECK_TOKEN;
+import static org.apache.pulsar.common.sasl.SaslConstants.SASL_TYPE_VALUE;
+import static org.assertj.core.api.Assertions.assertThat;
+import java.net.URI;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.AuthenticationDataProvider;
+import org.apache.pulsar.client.api.v5.auth.HttpAuthHeaders;
+import org.apache.pulsar.common.api.AuthData;
+import org.testng.annotations.Test;
+
+/**
+ * PIP-478 stage 3d: drive the real {@link SaslAuthenticationV5} SASL-over-HTTP body through the real
+ * {@link HttpAuthenticationDriver}, backed by a fake client-side SASL provider and a fake server transport
+ * that mimics {@code AuthenticationProviderSasl.authenticateHttpRequest}. This exercises the ported header
+ * state machine ({@code Init} → {@code ING} → {@code Done}, {@code SASL-Token} / {@code SASL-Server-ID} /
+ * role token) and, crucially, the state-slot persistence: one {@code PulsarSaslClient} conversation must
+ * span every round of one exchange and survive into the final role-token replay — no Kerberos or network.
+ */
+public class SaslAuthenticationV5HttpTest {
+
+ private static final URI ADMIN_URI = URI.create("https://broker.example:8443/admin/v2/tenants/x");
+
+ @Test
+ public void twoRoundExchangeReplaysRoleToken() throws Exception {
+ CountingFactory factory = new CountingFactory();
+ FakeSaslServer server = new FakeSaslServer(2, "role-token-abc");
+ HttpAuthenticationDriver driver =
+ new HttpAuthenticationDriver(new SaslAuthenticationV5(factory), null, server);
+
+ HttpAuthHeaders realRequestHeaders = driver.authenticateAsync(ADMIN_URI, null, Duration.ofSeconds(30)).get();
+
+ // The real request replays the validated role token with State=Done and the SASL type header.
+ assertThat(realRequestHeaders.get(SASL_AUTH_ROLE_TOKEN)).hasValue("role-token-abc");
+ assertThat(realRequestHeaders.get(SASL_HEADER_STATE)).hasValue(SASL_STATE_COMPLETE);
+ assertThat(realRequestHeaders.get(SASL_HEADER_TYPE)).hasValue(SASL_TYPE_VALUE);
+
+ // Exactly one PulsarSaslClient conversation was created for the whole exchange (state-slot persistence)
+ // and it advanced twice — the INIT frame plus one negotiation round.
+ assertThat(factory.createCalls).isEqualTo(1);
+ assertThat(factory.created.get(0).authenticateCalls).isEqualTo(2);
+
+ // The two warmup rounds were bodiless GETs to the original URI, carrying Init then ING.
+ assertThat(server.uris).hasSize(2).allMatch(ADMIN_URI::equals);
+ assertThat(server.requestStates).containsExactly(SASL_STATE_CLIENT_INIT, SASL_STATE_NEGOTIATE);
+
+ // Round 1 carried the initial SASL token; round 2 echoed the server id and carried the next token.
+ assertThat(server.requests.get(0).get(SASL_AUTH_TOKEN)).isPresent();
+ assertThat(server.requests.get(0).get(SASL_HEADER_TYPE)).hasValue(SASL_TYPE_VALUE);
+ assertThat(server.requests.get(1).get(SASL_STATE_SERVER)).hasValue("1");
+ assertThat(server.requests.get(1).get(SASL_AUTH_TOKEN)).isPresent();
+ }
+
+ @Test
+ public void singleRoundExchangeCompletesImmediately() throws Exception {
+ CountingFactory factory = new CountingFactory();
+ FakeSaslServer server = new FakeSaslServer(1, "rt-1");
+ HttpAuthenticationDriver driver =
+ new HttpAuthenticationDriver(new SaslAuthenticationV5(factory), null, server);
+
+ HttpAuthHeaders realRequestHeaders = driver.authenticateAsync(ADMIN_URI, null, Duration.ofSeconds(30)).get();
+
+ assertThat(realRequestHeaders.get(SASL_AUTH_ROLE_TOKEN)).hasValue("rt-1");
+ assertThat(realRequestHeaders.get(SASL_HEADER_STATE)).hasValue(SASL_STATE_COMPLETE);
+ assertThat(factory.createCalls).isEqualTo(1);
+ assertThat(factory.created.get(0).authenticateCalls).isEqualTo(1);
+ assertThat(server.requestStates).containsExactly(SASL_STATE_CLIENT_INIT);
+ }
+
+ @Test
+ public void secondRequestReplaysCachedRoleTokenWithServerCheckToken() throws Exception {
+ // PIP-478 FIX C: the framework HTTP driver reuses ONE body across requests. The first request runs a
+ // full negotiation and caches the role token on the body; the second request must replay that cached
+ // token with State=ServerCheckToken instead of restarting a fresh Kerberos Init negotiation (the v4
+ // AuthenticationSasl cross-request cache behavior).
+ CountingFactory factory = new CountingFactory();
+ RoleTokenServer server = new RoleTokenServer("role-token-abc");
+ HttpAuthenticationDriver driver =
+ new HttpAuthenticationDriver(new SaslAuthenticationV5(factory), null, server);
+
+ // First request: full negotiation issues and caches the role token.
+ HttpAuthHeaders first = driver.authenticateAsync(ADMIN_URI, null, Duration.ofSeconds(30)).get();
+ assertThat(first.get(SASL_AUTH_ROLE_TOKEN)).hasValue("role-token-abc");
+ assertThat(server.requestStates).containsExactly(SASL_STATE_CLIENT_INIT);
+
+ server.reset();
+
+ // Second request on the SAME driver/body: replayed with ServerCheckToken, NOT a fresh Init.
+ HttpAuthHeaders second = driver.authenticateAsync(ADMIN_URI, null, Duration.ofSeconds(30)).get();
+ assertThat(second.get(SASL_AUTH_ROLE_TOKEN)).hasValue("role-token-abc");
+ assertThat(second.get(SASL_HEADER_STATE)).hasValue(SASL_STATE_COMPLETE);
+ assertThat(server.requestStates).containsExactly(SASL_STATE_SERVER_CHECK_TOKEN);
+ // The single replayed warmup round carried the cached role token — no Init/negotiation round.
+ assertThat(server.requests.get(0).get(SASL_AUTH_ROLE_TOKEN)).hasValue("role-token-abc");
+ }
+
+ @Test
+ public void expiredCachedRoleTokenTriggersFreshNegotiation() throws Exception {
+ // PIP-478 FIX C: when the server reports the replayed cached token expired (SaslAuthRoleTokenExpired),
+ // the body drops the cache and renegotiates from Init, ending with a freshly issued role token.
+ CountingFactory factory = new CountingFactory();
+ RoleTokenServer server = new RoleTokenServer("role-token-1");
+ HttpAuthenticationDriver driver =
+ new HttpAuthenticationDriver(new SaslAuthenticationV5(factory), null, server);
+
+ // First request caches role-token-1.
+ driver.authenticateAsync(ADMIN_URI, null, Duration.ofSeconds(30)).get();
+
+ server.reset();
+ server.expireNextCheckThenIssue("role-token-2");
+
+ // Second request: ServerCheckToken -> server says EXPIRED -> renegotiate from Init -> role-token-2.
+ HttpAuthHeaders second = driver.authenticateAsync(ADMIN_URI, null, Duration.ofSeconds(30)).get();
+ assertThat(second.get(SASL_AUTH_ROLE_TOKEN)).hasValue("role-token-2");
+ assertThat(second.get(SASL_HEADER_STATE)).hasValue(SASL_STATE_COMPLETE);
+ assertThat(server.requestStates).containsExactly(SASL_STATE_SERVER_CHECK_TOKEN, SASL_STATE_CLIENT_INIT);
+ }
+
+ // ---- fakes ----
+
+ /** The client-side SASL data provider: deterministic tokens, no Kerberos; counts advances. */
+ private static final class FakeSaslClient implements AuthenticationDataProvider {
+ private static final long serialVersionUID = 1L;
+ int authenticateCalls;
+
+ @Override
+ public boolean hasDataForHttp() {
+ return true;
+ }
+
+ @Override
+ public Set<Entry<String, String>> getHttpHeaders() {
+ return Collections.singletonMap(SASL_HEADER_TYPE, SASL_TYPE_VALUE).entrySet();
+ }
+
+ @Override
+ public AuthData authenticate(AuthData commandData) {
+ authenticateCalls++;
+ return AuthData.of(("client-token-" + authenticateCalls).getBytes(UTF_8));
+ }
+ }
+
+ private static final class CountingFactory implements SaslAuthenticationV5.SaslProviderFactory {
+ private static final long serialVersionUID = 1L;
+ int createCalls;
+ final List<FakeSaslClient> created = new ArrayList<>();
+
+ @Override
+ public AuthenticationDataProvider create(String brokerHost) {
+ createCalls++;
+ FakeSaslClient client = new FakeSaslClient();
+ created.add(client);
+ return client;
+ }
+ }
+
+ /**
+ * A fake transport that mimics the broker's SASL-over-HTTP servlet: it {@code 401}s with a NEGOTIATE
+ * challenge until {@code roundsToComplete} client tokens have arrived, then {@code 200}s with a role
+ * token. Only the warmup rounds pass through here — the driver's final role-token request goes back to
+ * the caller, not the transport.
+ */
+ private static final class FakeSaslServer implements HttpChallengeTransport {
+ private final int roundsToComplete;
+ private final String issuedRoleToken;
+ private int serverStep;
+ final List<URI> uris = new ArrayList<>();
+ final List<String> requestStates = new ArrayList<>();
+ final List<HttpAuthHeaders> requests = new ArrayList<>();
+
+ FakeSaslServer(int roundsToComplete, String issuedRoleToken) {
+ this.roundsToComplete = roundsToComplete;
+ this.issuedRoleToken = issuedRoleToken;
+ }
+
+ @Override
+ public CompletableFuture<Result> get(URI uri, HttpAuthHeaders requestHeaders, Duration timeout) {
+ uris.add(uri);
+ requests.add(requestHeaders);
+ String state = requestHeaders.get(SASL_HEADER_STATE).orElse(null);
+ requestStates.add(state);
+ if (SASL_STATE_CLIENT_INIT.equalsIgnoreCase(state)) {
+ serverStep = 1;
+ } else if (SASL_STATE_NEGOTIATE.equalsIgnoreCase(state)) {
+ serverStep++;
+ } else {
+ return CompletableFuture.completedFuture(new Result(400, HttpAuthHeaders.empty()));
+ }
+ if (serverStep >= roundsToComplete) {
+ Map<String, String> headers = new LinkedHashMap<>();
+ headers.put(SASL_HEADER_TYPE, SASL_TYPE_VALUE);
+ headers.put(SASL_HEADER_STATE, SASL_STATE_COMPLETE);
+ headers.put(SASL_STATE_SERVER, "1");
+ headers.put(SASL_AUTH_ROLE_TOKEN, issuedRoleToken);
+ return CompletableFuture.completedFuture(new Result(200, HttpAuthHeaders.of(headers)));
+ }
+ Map<String, String> headers = new LinkedHashMap<>();
+ headers.put(SASL_HEADER_TYPE, SASL_TYPE_VALUE);
+ headers.put(SASL_HEADER_STATE, SASL_STATE_NEGOTIATE);
+ headers.put(SASL_STATE_SERVER, "1");
+ headers.put(SASL_AUTH_TOKEN, Base64.getEncoder().encodeToString(("server-" + serverStep).getBytes(UTF_8)));
+ return CompletableFuture.completedFuture(new Result(401, HttpAuthHeaders.of(headers)));
+ }
+ }
+
+ /**
+ * A fake server that also honors the role-token replay path (State=ServerCheckToken), used to exercise
+ * the FIX C cross-request cache. Init negotiation completes in one round. A ServerCheckToken is answered
+ * {@code 200} (validated) unless {@link #expireNextCheckThenIssue} armed an expiry, in which case the
+ * first ServerCheckToken is answered {@code 401} EXPIRED and the following Init issues the rotated token.
+ */
+ private static final class RoleTokenServer implements HttpChallengeTransport {
+ final List<String> requestStates = new ArrayList<>();
+ final List<HttpAuthHeaders> requests = new ArrayList<>();
+ private String roleToken;
+ private boolean expireNextCheck;
+ private String rotatedRoleToken;
+
+ RoleTokenServer(String roleToken) {
+ this.roleToken = roleToken;
+ }
+
+ void reset() {
+ requestStates.clear();
+ requests.clear();
+ }
+
+ void expireNextCheckThenIssue(String rotatedRoleToken) {
+ this.expireNextCheck = true;
+ this.rotatedRoleToken = rotatedRoleToken;
+ }
+
+ @Override
+ public CompletableFuture<Result> get(URI uri, HttpAuthHeaders requestHeaders, Duration timeout) {
+ requests.add(requestHeaders);
+ String state = requestHeaders.get(SASL_HEADER_STATE).orElse(null);
+ requestStates.add(state);
+ if (SASL_STATE_SERVER_CHECK_TOKEN.equalsIgnoreCase(state)) {
+ if (expireNextCheck) {
+ expireNextCheck = false;
+ Map<String, String> headers = new LinkedHashMap<>();
+ headers.put(SASL_HEADER_TYPE, SASL_TYPE_VALUE);
+ headers.put(SASL_HEADER_STATE, SASL_AUTH_ROLE_TOKEN_EXPIRED);
+ return CompletableFuture.completedFuture(new Result(401, HttpAuthHeaders.of(headers)));
+ }
+ return complete(roleToken);
+ }
+ if (SASL_STATE_CLIENT_INIT.equalsIgnoreCase(state)) {
+ if (rotatedRoleToken != null) {
+ roleToken = rotatedRoleToken;
+ rotatedRoleToken = null;
+ }
+ return complete(roleToken);
+ }
+ return CompletableFuture.completedFuture(new Result(400, HttpAuthHeaders.empty()));
+ }
+
+ private CompletableFuture<Result> complete(String issuedRoleToken) {
+ Map<String, String> headers = new LinkedHashMap<>();
+ headers.put(SASL_HEADER_TYPE, SASL_TYPE_VALUE);
+ headers.put(SASL_HEADER_STATE, SASL_STATE_COMPLETE);
+ headers.put(SASL_STATE_SERVER, "1");
+ headers.put(SASL_AUTH_ROLE_TOKEN, issuedRoleToken);
+ return CompletableFuture.completedFuture(new Result(200, HttpAuthHeaders.of(headers)));
+ }
+ }
+}