From fdc14d92b50c4f3324591dc015eb6f61a64d002d Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Mon, 10 Apr 2023 16:47:07 -0500 Subject: [PATCH 01/12] [improve][proxy] Only create ConnectionPool when needed --- .../pulsar/proxy/server/ProxyConnection.java | 45 ++++++++++--------- 1 file changed, 23 insertions(+), 22 deletions(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 9530389b524b3..16be3d9bf7d14 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -81,7 +81,7 @@ */ public class ProxyConnection extends PulsarHandler { private static final Logger LOG = LoggerFactory.getLogger(ProxyConnection.class); - // ConnectionPool is used by the proxy to issue lookup requests + // ConnectionPool is used by the proxy to issue lookup requests. It is null when doing direct broker proxying. private ConnectionPool connectionPool; private final AtomicLong requestIdGenerator = new AtomicLong(ThreadLocalRandom.current().nextLong(0, Long.MAX_VALUE / 2)); @@ -313,24 +313,7 @@ protected static boolean isTlsChannel(Channel channel) { } private synchronized void completeConnect() throws PulsarClientException { - Supplier clientCnxSupplier; - if (service.getConfiguration().isAuthenticationEnabled()) { - clientCnxSupplier = () -> new ProxyClientCnx(clientConf, service.getWorkerGroup(), clientAuthRole, - clientAuthData, clientAuthMethod, protocolVersionToAdvertise, - service.getConfiguration().isForwardAuthorizationCredentials(), this); - } else { - clientCnxSupplier = () -> new ClientCnx(clientConf, service.getWorkerGroup(), protocolVersionToAdvertise); - } - - if (this.connectionPool == null) { - this.connectionPool = new ConnectionPool(clientConf, service.getWorkerGroup(), - clientCnxSupplier, - Optional.of(dnsAddressResolverGroup.getResolver(service.getWorkerGroup().next()))); - } else { - LOG.error("BUG! Connection Pool has already been created for proxy connection to {} state {} role {}", - remoteAddress, state, clientAuthRole); - } - + checkArgument(state == State.Connecting); LOG.info("[{}] complete connection, init proxy handler. authenticated with {} role {}, hasProxyToBrokerUrl: {}", remoteAddress, authMethod, clientAuthRole, hasProxyToBrokerUrl); if (hasProxyToBrokerUrl) { @@ -371,8 +354,26 @@ private synchronized void completeConnect() throws PulsarClientException { }); } else { // Client is doing a lookup, we can consider the handshake complete - // and we'll take care of just topics and - // partitions metadata lookups + // and we'll take care of just topics and partitions metadata lookups + Supplier clientCnxSupplier; + if (service.getConfiguration().isAuthenticationEnabled()) { + clientCnxSupplier = () -> new ProxyClientCnx(clientConf, service.getWorkerGroup(), clientAuthRole, + clientAuthData, clientAuthMethod, protocolVersionToAdvertise, + service.getConfiguration().isForwardAuthorizationCredentials(), this); + } else { + clientCnxSupplier = + () -> new ClientCnx(clientConf, service.getWorkerGroup(), protocolVersionToAdvertise); + } + + if (this.connectionPool == null) { + this.connectionPool = new ConnectionPool(clientConf, service.getWorkerGroup(), + clientCnxSupplier, + Optional.of(dnsAddressResolverGroup.getResolver(service.getWorkerGroup().next()))); + } else { + LOG.error("BUG! Connection Pool has already been created for proxy connection to {} state {} role {}", + remoteAddress, state, clientAuthRole); + } + state = State.ProxyLookupRequests; lookupProxyHandler = new LookupProxyHandler(service, this); final ByteBuf msg = Commands.newConnected(protocolVersionToAdvertise, false); @@ -452,7 +453,7 @@ protected void authChallengeSuccessCallback(AuthData authChallenge) { } // First connection - if (this.connectionPool == null || state == State.Connecting) { + if (state == State.Connecting) { // authentication has completed, will send newConnected command. completeConnect(); } From 23097511f2dafbfb19debc92f66eeca89a0f2291 Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Mon, 10 Apr 2023 22:26:35 -0500 Subject: [PATCH 02/12] [fix][proxy] Refresh auth data if ProxyLookupRequests --- .../pulsar/proxy/server/ProxyClientCnx.java | 63 +++---- .../proxy/server/ProxyConfiguration.java | 7 + .../pulsar/proxy/server/ProxyConnection.java | 158 ++++++++++++------ .../proxy/server/ProxyRefreshAuthTest.java | 1 + 4 files changed, 140 insertions(+), 89 deletions(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java index a1994fb5af4b0..64df673a9eaa2 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java @@ -23,15 +23,17 @@ import io.netty.channel.EventLoopGroup; import java.util.Arrays; import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.PulsarVersion; import org.apache.pulsar.client.impl.ClientCnx; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.apache.pulsar.common.api.AuthData; import org.apache.pulsar.common.api.proto.CommandAuthChallenge; import org.apache.pulsar.common.protocol.Commands; +import org.apache.pulsar.common.util.netty.NettyChannelUtil; @Slf4j /** - * Channel handler for Pulsar proxy's Pulsar broker client connections. + * Channel handler for Pulsar proxy's Pulsar broker client connections for lookup requests. *

* Please see {@link org.apache.pulsar.common.protocol.PulsarDecoder} javadoc for important details about handle* * method parameter instance lifecycle. @@ -40,15 +42,13 @@ public class ProxyClientCnx extends ClientCnx { private final boolean forwardClientAuthData; private final String clientAuthMethod; private final String clientAuthRole; - private final AuthData clientAuthData; private final ProxyConnection proxyConnection; public ProxyClientCnx(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, String clientAuthRole, - AuthData clientAuthData, String clientAuthMethod, int protocolVersion, + String clientAuthMethod, int protocolVersion, boolean forwardClientAuthData, ProxyConnection proxyConnection) { super(conf, eventLoopGroup, protocolVersion); this.clientAuthRole = clientAuthRole; - this.clientAuthData = clientAuthData; this.clientAuthMethod = clientAuthMethod; this.forwardClientAuthData = forwardClientAuthData; this.proxyConnection = proxyConnection; @@ -59,9 +59,15 @@ protected ByteBuf newConnectCommand() throws Exception { if (log.isDebugEnabled()) { log.debug("New Connection opened via ProxyClientCnx with params clientAuthRole = {}," + " clientAuthData = {}, clientAuthMethod = {}", - clientAuthRole, clientAuthData, clientAuthMethod); + clientAuthRole, proxyConnection.getClientAuthData(), clientAuthMethod); + } + AuthData clientAuthData = null; + if (forwardClientAuthData) { + // There is a chance this auth data is expired because the ProxyConnection does not do early token refresh. + // Based on the current design, the best option is to configure the broker to accept slightly stale + // authentication data. + clientAuthData = proxyConnection.getClientAuthData(); } - authenticationDataProvider = authentication.getAuthData(remoteHostName); AuthData authData = authenticationDataProvider.authenticate(AuthData.INIT_AUTH_DATA); return Commands.newConnect(authentication.getAuthMethodName(), authData, protocolVersion, @@ -75,43 +81,16 @@ protected void handleAuthChallenge(CommandAuthChallenge authChallenge) { checkArgument(authChallenge.getChallenge().hasAuthData()); boolean isRefresh = Arrays.equals(AuthData.REFRESH_AUTH_DATA_BYTES, authChallenge.getChallenge().getAuthData()); - if (!forwardClientAuthData || !isRefresh) { + if (forwardClientAuthData && isRefresh) { + proxyConnection.getValidClientAuthData() + .thenApplyAsync(authData -> { + NettyChannelUtil.writeAndFlushWithVoidPromise(ctx, + Commands.newAuthResponse(clientAuthMethod, authData, this.protocolVersion, + String.format("Pulsar-Java-v%s", PulsarVersion.getVersion()))); + return null; + }, ctx.executor()); + } else { super.handleAuthChallenge(authChallenge); - return; - } - - try { - if (log.isDebugEnabled()) { - log.debug("Proxy {} request to refresh the original client authentication data for " - + "the proxy client {}", proxyConnection.ctx().channel(), ctx.channel()); - } - - proxyConnection.ctx().writeAndFlush(Commands.newAuthChallenge(clientAuthMethod, AuthData.REFRESH_AUTH_DATA, - protocolVersion)) - .addListener(writeFuture -> { - if (writeFuture.isSuccess()) { - if (log.isDebugEnabled()) { - log.debug("Proxy {} sent the auth challenge to original client to refresh credentials " - + "with method {} for the proxy client {}", - proxyConnection.ctx().channel(), clientAuthMethod, ctx.channel()); - } - } else { - log.error("Failed to send the auth challenge to original client by the proxy {} " - + "for the proxy client {}", - proxyConnection.ctx().channel(), - ctx.channel(), - writeFuture.cause()); - closeWithException(writeFuture.cause()); - } - }); - - if (state == State.SentConnectFrame) { - state = State.Connecting; - } - } catch (Exception e) { - log.error("Failed to send the auth challenge to origin client by the proxy {} for the proxy client {}", - proxyConnection.ctx().channel(), ctx.channel(), e); - closeWithException(e); } } } diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java index a91b6e70f5b8b..a6aac6be966fd 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java @@ -364,6 +364,13 @@ public class ProxyConfiguration implements PulsarConfiguration { + "to take effect" ) private boolean forwardAuthorizationCredentials = false; + + @FieldContext( + category = CATEGORY_AUTHENTICATION, + doc = "Interval of time for checking for expired authentication credentials. Disable by setting to 0." + ) + private int authenticationRefreshCheckSeconds = 60; + @FieldContext( category = CATEGORY_AUTHENTICATION, doc = "Whether the '/metrics' endpoint requires authentication. Defaults to true." diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 16be3d9bf7d14..dda6303537083 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -29,13 +29,17 @@ import io.netty.handler.codec.haproxy.HAProxyMessage; import io.netty.handler.ssl.SslHandler; import io.netty.resolver.dns.DnsAddressResolverGroup; +import io.netty.util.concurrent.ScheduledFuture; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.nio.channels.ClosedChannelException; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; @@ -64,6 +68,7 @@ import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace; import org.apache.pulsar.common.api.proto.CommandLookupTopic; import org.apache.pulsar.common.api.proto.CommandPartitionedTopicMetadata; +import org.apache.pulsar.common.api.proto.FeatureFlags; import org.apache.pulsar.common.api.proto.ProtocolVersion; import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.protocol.Commands; @@ -92,10 +97,14 @@ public class ProxyConnection extends PulsarHandler { private LookupProxyHandler lookupProxyHandler = null; @Getter private DirectProxyHandler directProxyHandler = null; + private ScheduledFuture authRefreshTask; + private long authChallengeSentTime = Long.MAX_VALUE; + private FeatureFlags features; + private Set> pendingBrokerAuthChallenges = null; private final BrokerProxyValidator brokerProxyValidator; private final ConnectionController connectionController; String clientAuthRole; - AuthData clientAuthData; + volatile AuthData clientAuthData; String clientAuthMethod; String clientVersion; @@ -191,6 +200,15 @@ public synchronized void channelInactive(ChannelHandlerContext ctx) throws Excep directProxyHandler = null; } + if (authRefreshTask != null) { + authRefreshTask.cancel(false); + } + + if (pendingBrokerAuthChallenges != null) { + pendingBrokerAuthChallenges.forEach(future -> future.cancel(true)); + pendingBrokerAuthChallenges = null; + } + service.getClientCnxs().remove(this); LOG.info("[{}] Connection closed", remoteAddress); @@ -358,7 +376,7 @@ private synchronized void completeConnect() throws PulsarClientException { Supplier clientCnxSupplier; if (service.getConfiguration().isAuthenticationEnabled()) { clientCnxSupplier = () -> new ProxyClientCnx(clientConf, service.getWorkerGroup(), clientAuthRole, - clientAuthData, clientAuthMethod, protocolVersionToAdvertise, + clientAuthMethod, protocolVersionToAdvertise, service.getConfiguration().isForwardAuthorizationCredentials(), this); } else { clientCnxSupplier = @@ -376,6 +394,13 @@ private synchronized void completeConnect() throws PulsarClientException { state = State.ProxyLookupRequests; lookupProxyHandler = new LookupProxyHandler(service, this); + if (service.getConfiguration().isAuthenticationEnabled() + && service.getConfiguration().getAuthenticationRefreshCheckSeconds() > 0) { + authRefreshTask = ctx.executor().scheduleAtFixedRate(() -> refreshAuthenticationCredentials(false), + service.getConfiguration().getAuthenticationRefreshCheckSeconds(), + service.getConfiguration().getAuthenticationRefreshCheckSeconds(), + TimeUnit.SECONDS); + } final ByteBuf msg = Commands.newConnected(protocolVersionToAdvertise, false); writeAndFlush(msg); } @@ -472,6 +497,47 @@ protected void authChallengeSuccessCallback(AuthData authChallenge) { } } + private void refreshAuthenticationCredentials(boolean force) { + assert ctx.executor().inEventLoop(); + if (state != State.ProxyLookupRequests) { + // Happens when an exception is thrown that causes this connection to close. + return; + } else if (!authState.isExpired() || !force) { + // Credentials are still valid. Nothing to do at this point + return; + } + + if (!supportsAuthenticationRefresh()) { + LOG.warn("[{}] Closing connection because client doesn't support auth credentials refresh", remoteAddress); + ctx.close(); + return; + } + + if (System.nanoTime() - authChallengeSentTime + > TimeUnit.SECONDS.toNanos(service.getConfiguration().getAuthenticationRefreshCheckSeconds())) { + LOG.warn("[{}] Closing connection after timeout on refreshing auth credentials", remoteAddress); + ctx.close(); + return; + } + + if (LOG.isDebugEnabled()) { + LOG.info("[{}] Refreshing authentication credentials", remoteAddress); + } + + try { + AuthData brokerData = authState.refreshAuthentication(); + writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, getRemoteEndpointProtocolVersion())); + if (LOG.isDebugEnabled()) { + LOG.debug("[{}] Sent auth challenge to client to refresh credentials with method: {}.", + remoteAddress, authMethod); + } + authChallengeSentTime = System.nanoTime(); + } catch (AuthenticationException e) { + LOG.warn("[{}] Failed to refresh authentication: {}", remoteAddress, e); + ctx.close(); + } + } + @Override protected void handleConnect(CommandConnect connect) { checkArgument(state == State.Init); @@ -481,6 +547,10 @@ protected void handleConnect(CommandConnect connect) { this.protocolVersionToAdvertise = getProtocolVersionToAdvertise(connect); this.proxyToBrokerUrl = connect.hasProxyToBrokerUrl() ? connect.getProxyToBrokerUrl() : "null"; this.clientVersion = connect.getClientVersion(); + features = new FeatureFlags(); + if (connect.hasFeatureFlags()) { + features.copyFrom(connect.getFeatureFlags()); + } if (LOG.isDebugEnabled()) { LOG.debug("Received CONNECT from {} proxyToBroker={}", remoteAddress, proxyToBrokerUrl); @@ -565,53 +635,19 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) { } try { + authChallengeSentTime = Long.MAX_VALUE; AuthData clientData = AuthData.of(authResponse.getResponse().getAuthData()); + // Authenticate the client's auth data and send to the broker concurrently + // Note: this implementation relies on the current weakness that prevents multi-stage authentication + // from working when forwardAuthorizationCredentials is enabled. Here is an issue to fix the protocol: + // https://github.com/apache/pulsar/issues/19291. doAuthentication(clientData); - if (service.getConfiguration().isForwardAuthorizationCredentials() - && connectionPool != null && state == State.ProxyLookupRequests) { - connectionPool.getConnections().forEach(toBrokerCnxFuture -> { - String clientVersion; - if (authResponse.hasClientVersion()) { - clientVersion = authResponse.getClientVersion(); - } else { - clientVersion = this.clientVersion; - } - int protocolVersion; - if (authResponse.hasProtocolVersion()) { - protocolVersion = authResponse.getProtocolVersion(); - } else { - protocolVersion = Commands.getCurrentProtocolVersion(); - } - - ByteBuf cmd = - Commands.newAuthResponse(clientAuthMethod, clientData, protocolVersion, clientVersion); - toBrokerCnxFuture.thenAccept(toBrokerCnx -> toBrokerCnx.ctx().writeAndFlush(cmd) - .addListener(writeFuture -> { - if (writeFuture.isSuccess()) { - if (LOG.isDebugEnabled()) { - LOG.debug("{} authentication is refreshed successfully by {}, " - + "auth method: {} ", - toBrokerCnx.ctx().channel(), ctx.channel(), clientAuthMethod); - } - } else { - LOG.error("Failed to forward the auth response " - + "from the proxy to the broker through the proxy client, " - + "proxy: {}, proxy client: {}", - ctx.channel(), - toBrokerCnx.ctx().channel(), - writeFuture.cause()); - toBrokerCnx.ctx().channel().pipeline() - .fireExceptionCaught(writeFuture.cause()); - } - })) - .whenComplete((__, ex) -> { - if (ex != null) { - LOG.error("Failed to forward the auth response from the proxy to " - + "the broker through the proxy client, proxy: {}", - ctx().channel(), ex); - } - }); - }); + if (pendingBrokerAuthChallenges != null && !pendingBrokerAuthChallenges.isEmpty()) { + // Send pending auth data requests to the broker + for (CompletableFuture challenge : pendingBrokerAuthChallenges) { + challenge.complete(clientData); + } + pendingBrokerAuthChallenges.clear(); } } catch (Exception e) { String errorMsg = "Unable to handleAuthResponse"; @@ -760,4 +796,32 @@ private void writeAndFlush(ByteBuf cmd) { private void writeAndFlushAndClose(ByteBuf cmd) { NettyChannelUtil.writeAndFlushWithClosePromise(ctx, cmd); } + + boolean supportsAuthenticationRefresh() { + return features != null && features.isSupportsAuthRefresh(); + } + + AuthData getClientAuthData() { + return clientAuthData; + } + + CompletableFuture getValidClientAuthData() { + final CompletableFuture clientAuthDataFuture = new CompletableFuture<>(); + ctx().executor().execute(() -> { + // authState is not thread safe, so this must run on the ProxyConnection's event loop. + if (!authState.isExpired()) { + clientAuthDataFuture.complete(clientAuthData); + } else { + if (authChallengeSentTime == Long.MAX_VALUE) { + // We only need to issue an auth challenge if we are not waiting on a response from the client. + refreshAuthenticationCredentials(true); + } + if (pendingBrokerAuthChallenges == null) { + pendingBrokerAuthChallenges = new HashSet<>(); + } + pendingBrokerAuthChallenges.add(clientAuthDataFuture); + } + }); + return clientAuthDataFuture; + } } diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java index d14105b0b43c2..db064e35e63d5 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java @@ -108,6 +108,7 @@ protected void setup() throws Exception { proxyConfig.setAuthenticationEnabled(true); proxyConfig.setAuthorizationEnabled(false); proxyConfig.setForwardAuthorizationCredentials(true); + proxyConfig.setAuthenticationRefreshCheckSeconds(1); proxyConfig.setBrokerServiceURL(pulsar.getBrokerServiceUrl()); proxyConfig.setAdvertisedAddress(null); From 243dc01d41f7ee3caaa3b31c31630fd0006c1cc4 Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Tue, 11 Apr 2023 00:04:17 -0500 Subject: [PATCH 03/12] Add better test; support allowed skew for token auth --- .../AuthenticationProviderToken.java | 22 ++++++++++++++- .../pulsar/proxy/server/ProxyConnection.java | 3 ++ .../proxy/server/ProxyRefreshAuthTest.java | 28 ++++++++++++++----- 3 files changed, 45 insertions(+), 8 deletions(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java index fed5ba063fd44..67e1f39a38924 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java @@ -73,6 +73,9 @@ public class AuthenticationProviderToken implements AuthenticationProvider { // The token audience stands for this broker. The field `tokenAudienceClaim` of a valid token, need contains this. static final String CONF_TOKEN_AUDIENCE = "tokenAudience"; + // The amount of time in seconds that a token is allowed to be out of sync with the server's time when performing + // token validation. + static final String CONF_TOKEN_ALLOWED_CLOCK_SKEW_SECONDS = "tokenAllowedClockSkewSeconds"; static final String TOKEN = "token"; @@ -101,6 +104,7 @@ public class AuthenticationProviderToken implements AuthenticationProvider { private String confTokenPublicAlgSettingName; private String confTokenAudienceClaimSettingName; private String confTokenAudienceSettingName; + private String confTokenAllowedClockSkewSecondsSettingName; @Override public void close() throws IOException { @@ -125,6 +129,7 @@ public void initialize(ServiceConfiguration config) throws IOException, IllegalA this.confTokenPublicAlgSettingName = prefix + CONF_TOKEN_PUBLIC_ALG; this.confTokenAudienceClaimSettingName = prefix + CONF_TOKEN_AUDIENCE_CLAIM; this.confTokenAudienceSettingName = prefix + CONF_TOKEN_AUDIENCE; + this.confTokenAllowedClockSkewSecondsSettingName = prefix + CONF_TOKEN_ALLOWED_CLOCK_SKEW_SECONDS; // we need to fetch the algorithm before we fetch the key this.publicKeyAlg = getPublicKeyAlgType(config); @@ -133,7 +138,12 @@ public void initialize(ServiceConfiguration config) throws IOException, IllegalA this.audienceClaim = getTokenAudienceClaim(config); this.audience = getTokenAudience(config); - this.parser = Jwts.parserBuilder().setSigningKey(this.validationKey).build(); + long allowedSkew = getConfTokenAllowedClockSkewSeconds(config); + + this.parser = Jwts.parserBuilder() + .setAllowedClockSkewSeconds(allowedSkew) + .setSigningKey(this.validationKey) + .build(); if (audienceClaim != null && audience == null) { throw new IllegalArgumentException("Token Audience Claim [" + audienceClaim @@ -329,6 +339,16 @@ private String getTokenAudience(ServiceConfiguration conf) throws IllegalArgumen } } + // get Token's allowed clock skew in seconds. If not configured, defaults to 0. + private long getConfTokenAllowedClockSkewSeconds(ServiceConfiguration conf) throws IllegalArgumentException { + String allowedSkewStr = (String) conf.getProperty(confTokenAllowedClockSkewSecondsSettingName); + if (StringUtils.isNotBlank(allowedSkewStr)) { + return Long.parseLong(allowedSkewStr); + } else { + return 0; + } + } + private static final class TokenAuthenticationState implements AuthenticationState { private final AuthenticationProviderToken provider; private final SocketAddress remoteAddress; diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index a3368dfb179c9..42b50be4d5865 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -642,7 +642,10 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) { // from working when forwardAuthorizationCredentials is enabled. Here is an issue to fix the protocol: // https://github.com/apache/pulsar/issues/19291. doAuthentication(clientData); + // We only have pendingBrokerAuthChallenges when forwardAuthorizationCredentials is enabled. if (pendingBrokerAuthChallenges != null && !pendingBrokerAuthChallenges.isEmpty()) { + // Must store the clientAuthData to be able to initialize future ProxyClientCnx. + this.clientAuthData = clientData; // Send pending auth data requests to the broker for (CompletableFuture challenge : pendingBrokerAuthChallenges) { challenge.complete(clientData); diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java index db064e35e63d5..d442e32c2f2f8 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java @@ -19,7 +19,6 @@ package org.apache.pulsar.proxy.server; import static java.util.concurrent.TimeUnit.SECONDS; -import static org.mockito.Mockito.spy; import static org.testng.Assert.assertTrue; import com.google.common.collect.Sets; import io.jsonwebtoken.SignatureAlgorithm; @@ -31,13 +30,11 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import javax.crypto.SecretKey; -import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.authentication.AuthenticationProviderToken; import org.apache.pulsar.broker.authentication.AuthenticationService; import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils; import org.apache.pulsar.client.admin.PulsarAdmin; -import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerConsumerBase; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.impl.ClientCnx; @@ -81,6 +78,7 @@ protected void doInitConf() throws Exception { conf.setAuthenticationProviders(Set.of(AuthenticationProviderToken.class.getName())); Properties properties = new Properties(); properties.setProperty("tokenSecretKey", AuthTokenUtils.encodeKeyBase64(SECRET_KEY)); + properties.setProperty("tokenAllowedClockSkewSeconds", "3"); conf.setProperties(properties); conf.setClusterName("proxy-authorization"); @@ -163,14 +161,30 @@ public void testAuthDataRefresh(boolean forwardAuthData) throws Exception { .authentication(authenticationToken)); String topic = "persistent://my-tenant/my-ns/my-topic1"; - @Cleanup - Producer ignored = spy(pulsarClient.newProducer() - .topic(topic).create()); PulsarClientImpl pulsarClientImpl = (PulsarClientImpl) pulsarClient; + pulsarClient.getPartitionsForTopic(topic).get(); Set> connections = pulsarClientImpl.getCnxPool().getConnections(); - Awaitility.await().during(4, SECONDS).untilAsserted(() -> { + Awaitility.await().during(5, SECONDS).untilAsserted(() -> { + pulsarClient.getPartitionsForTopic(topic).get(); + assertTrue(connections.stream().allMatch(n -> { + try { + ClientCnx clientCnx = n.get(); + long timestamp = clientCnx.getLastDisconnectedTimestamp(); + return timestamp == 0; + } catch (Exception e) { + throw new RuntimeException(e); + } + })); + }); + + // Force all connections from proxy to broker to close and therefore require the proxy to re-authenticate with + // the broker. (The client doesn't lose this connection.) + restartBroker(); + + // Rerun assertion to ensure that it still works + Awaitility.await().during(5, SECONDS).untilAsserted(() -> { pulsarClient.getPartitionsForTopic(topic).get(); assertTrue(connections.stream().allMatch(n -> { try { From b5872689cbceff68d1d060c82b13b1618a50cc37 Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Tue, 11 Apr 2023 08:06:56 -0500 Subject: [PATCH 04/12] Fix log level; use catchingAndLoggingThrowables --- .../apache/pulsar/proxy/server/ProxyConnection.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 42b50be4d5865..b3792bf459056 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -73,6 +73,7 @@ import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.protocol.PulsarHandler; +import org.apache.pulsar.common.util.Runnables; import org.apache.pulsar.common.util.netty.NettyChannelUtil; import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData; import org.slf4j.Logger; @@ -396,7 +397,8 @@ private synchronized void completeConnect() throws PulsarClientException { lookupProxyHandler = service.newLookupProxyHandler(this); if (service.getConfiguration().isAuthenticationEnabled() && service.getConfiguration().getAuthenticationRefreshCheckSeconds() > 0) { - authRefreshTask = ctx.executor().scheduleAtFixedRate(() -> refreshAuthenticationCredentials(false), + authRefreshTask = ctx.executor().scheduleAtFixedRate( + Runnables.catchingAndLoggingThrowables(() -> refreshAuthenticationCredentials(false)), service.getConfiguration().getAuthenticationRefreshCheckSeconds(), service.getConfiguration().getAuthenticationRefreshCheckSeconds(), TimeUnit.SECONDS); @@ -521,7 +523,7 @@ private void refreshAuthenticationCredentials(boolean force) { } if (LOG.isDebugEnabled()) { - LOG.info("[{}] Refreshing authentication credentials", remoteAddress); + LOG.debug("[{}] Refreshing authentication credentials", remoteAddress); } try { @@ -810,7 +812,7 @@ AuthData getClientAuthData() { CompletableFuture getValidClientAuthData() { final CompletableFuture clientAuthDataFuture = new CompletableFuture<>(); - ctx().executor().execute(() -> { + ctx().executor().execute(Runnables.catchingAndLoggingThrowables(() ->{ // authState is not thread safe, so this must run on the ProxyConnection's event loop. if (!authState.isExpired()) { clientAuthDataFuture.complete(clientAuthData); @@ -824,7 +826,7 @@ CompletableFuture getValidClientAuthData() { } pendingBrokerAuthChallenges.add(clientAuthDataFuture); } - }); + })); return clientAuthDataFuture; } } From 0a6be923716b10833fe5db0f20e35cabd059182f Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Tue, 11 Apr 2023 08:30:39 -0500 Subject: [PATCH 05/12] Add check for proxy state to ensure correctness --- .../org/apache/pulsar/proxy/server/ProxyClientCnx.java | 7 ++++++- .../org/apache/pulsar/proxy/server/ProxyConnection.java | 7 ++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java index 64df673a9eaa2..2760692f89ed0 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java @@ -88,7 +88,12 @@ protected void handleAuthChallenge(CommandAuthChallenge authChallenge) { Commands.newAuthResponse(clientAuthMethod, authData, this.protocolVersion, String.format("Pulsar-Java-v%s", PulsarVersion.getVersion()))); return null; - }, ctx.executor()); + }, ctx.executor()) + .exceptionally(ex -> { + log.warn("Failed to get valid client auth data", ex); + ctx.close(); + return null; + }); } else { super.handleAuthChallenge(authChallenge); } diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index b3792bf459056..7b381acb82a3c 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -812,7 +812,12 @@ AuthData getClientAuthData() { CompletableFuture getValidClientAuthData() { final CompletableFuture clientAuthDataFuture = new CompletableFuture<>(); - ctx().executor().execute(Runnables.catchingAndLoggingThrowables(() ->{ + ctx().executor().execute(Runnables.catchingAndLoggingThrowables(() -> { + if (state != State.ProxyLookupRequests) { + clientAuthDataFuture.completeExceptionally(new PulsarClientException.AlreadyClosedException( + "ProxyConnection is not in a valid state to get client auth data")); + return; + } // authState is not thread safe, so this must run on the ProxyConnection's event loop. if (!authState.isExpired()) { clientAuthDataFuture.complete(clientAuthData); From bfc4efa1eb63adf77272b1628615253c4a1c1429 Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Tue, 11 Apr 2023 08:31:57 -0500 Subject: [PATCH 06/12] Make exception log more unique --- .../java/org/apache/pulsar/proxy/server/ProxyConnection.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 7b381acb82a3c..9719d78eb8cfa 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -815,7 +815,7 @@ CompletableFuture getValidClientAuthData() { ctx().executor().execute(Runnables.catchingAndLoggingThrowables(() -> { if (state != State.ProxyLookupRequests) { clientAuthDataFuture.completeExceptionally(new PulsarClientException.AlreadyClosedException( - "ProxyConnection is not in a valid state to get client auth data")); + "ProxyConnection is not in a valid state to get client auth data for " + remoteAddress)); return; } // authState is not thread safe, so this must run on the ProxyConnection's event loop. From 51ec3f4822a954d65d042e5694f901e5bade987b Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Tue, 11 Apr 2023 11:15:02 -0500 Subject: [PATCH 07/12] Fix refresh logic --- .../pulsar/proxy/server/ProxyClientCnx.java | 2 +- .../pulsar/proxy/server/ProxyConnection.java | 48 +++++++++++-------- 2 files changed, 29 insertions(+), 21 deletions(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java index 2760692f89ed0..6c1564281addd 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java @@ -90,7 +90,7 @@ protected void handleAuthChallenge(CommandAuthChallenge authChallenge) { return null; }, ctx.executor()) .exceptionally(ex -> { - log.warn("Failed to get valid client auth data", ex); + log.warn("Failed to get valid client auth data. Closing connection.", ex); ctx.close(); return null; }); diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 9719d78eb8cfa..3c5b309b258a9 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -99,6 +99,8 @@ public class ProxyConnection extends PulsarHandler { @Getter private DirectProxyHandler directProxyHandler = null; private ScheduledFuture authRefreshTask; + // When authChallengeSentTime is not Long.MAX_VALUE, it means the proxy is waiting for the client to respond + // to an auth challenge. When authChallengeSentTime is Long.MAX_VALUE, there are no pending auth challenges. private long authChallengeSentTime = Long.MAX_VALUE; private FeatureFlags features; private Set> pendingBrokerAuthChallenges = null; @@ -398,7 +400,7 @@ private synchronized void completeConnect() throws PulsarClientException { if (service.getConfiguration().isAuthenticationEnabled() && service.getConfiguration().getAuthenticationRefreshCheckSeconds() > 0) { authRefreshTask = ctx.executor().scheduleAtFixedRate( - Runnables.catchingAndLoggingThrowables(() -> refreshAuthenticationCredentials(false)), + Runnables.catchingAndLoggingThrowables(this::refreshAuthenticationCredentialsAndCloseIfTooExpired), service.getConfiguration().getAuthenticationRefreshCheckSeconds(), service.getConfiguration().getAuthenticationRefreshCheckSeconds(), TimeUnit.SECONDS); @@ -499,33 +501,40 @@ protected void authChallengeSuccessCallback(AuthData authChallenge) { } } - private void refreshAuthenticationCredentials(boolean force) { + private void refreshAuthenticationCredentialsAndCloseIfTooExpired() { assert ctx.executor().inEventLoop(); if (state != State.ProxyLookupRequests) { // Happens when an exception is thrown that causes this connection to close. return; - } else if (!authState.isExpired() || !force) { + } else if (!authState.isExpired()) { // Credentials are still valid. Nothing to do at this point return; } - if (!supportsAuthenticationRefresh()) { - LOG.warn("[{}] Closing connection because client doesn't support auth credentials refresh", remoteAddress); - ctx.close(); - return; - } - if (System.nanoTime() - authChallengeSentTime > TimeUnit.SECONDS.toNanos(service.getConfiguration().getAuthenticationRefreshCheckSeconds())) { LOG.warn("[{}] Closing connection after timeout on refreshing auth credentials", remoteAddress); ctx.close(); + } + + maybeSendAuthChallenge(); + } + + private void maybeSendAuthChallenge() { + assert ctx.executor().inEventLoop(); + + if (!supportsAuthenticationRefresh()) { + LOG.warn("[{}] Closing connection because client doesn't support auth credentials refresh", remoteAddress); + ctx.close(); + return; + } else if (authChallengeSentTime == Long.MAX_VALUE) { + // If the proxy sent a refresh but hasn't yet heard back, do not send another challenge. return; } if (LOG.isDebugEnabled()) { LOG.debug("[{}] Refreshing authentication credentials", remoteAddress); } - try { AuthData brokerData = authState.refreshAuthentication(); writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, getRemoteEndpointProtocolVersion())); @@ -810,26 +819,25 @@ AuthData getClientAuthData() { return clientAuthData; } + /** + * Thread-safe method to retrieve unexpired client auth data. Due to inherent race conditions, + * the auth data may expire before it is used. + */ CompletableFuture getValidClientAuthData() { final CompletableFuture clientAuthDataFuture = new CompletableFuture<>(); ctx().executor().execute(Runnables.catchingAndLoggingThrowables(() -> { - if (state != State.ProxyLookupRequests) { - clientAuthDataFuture.completeExceptionally(new PulsarClientException.AlreadyClosedException( - "ProxyConnection is not in a valid state to get client auth data for " + remoteAddress)); - return; - } // authState is not thread safe, so this must run on the ProxyConnection's event loop. if (!authState.isExpired()) { clientAuthDataFuture.complete(clientAuthData); - } else { - if (authChallengeSentTime == Long.MAX_VALUE) { - // We only need to issue an auth challenge if we are not waiting on a response from the client. - refreshAuthenticationCredentials(true); - } + } else if (state == State.ProxyLookupRequests) { + maybeSendAuthChallenge(); if (pendingBrokerAuthChallenges == null) { pendingBrokerAuthChallenges = new HashSet<>(); } pendingBrokerAuthChallenges.add(clientAuthDataFuture); + } else { + clientAuthDataFuture.completeExceptionally(new PulsarClientException.AlreadyClosedException( + "ProxyConnection is not in a valid state to get client auth data for " + remoteAddress)); } })); return clientAuthDataFuture; From 39a8cf246b51fd570b656bc1e300f60f8b863046 Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Tue, 11 Apr 2023 11:48:26 -0500 Subject: [PATCH 08/12] Fix checkstyle --- .../java/org/apache/pulsar/proxy/server/ProxyConnection.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 3c5b309b258a9..59eb3b87d7db5 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -400,7 +400,8 @@ private synchronized void completeConnect() throws PulsarClientException { if (service.getConfiguration().isAuthenticationEnabled() && service.getConfiguration().getAuthenticationRefreshCheckSeconds() > 0) { authRefreshTask = ctx.executor().scheduleAtFixedRate( - Runnables.catchingAndLoggingThrowables(this::refreshAuthenticationCredentialsAndCloseIfTooExpired), + Runnables.catchingAndLoggingThrowables( + this::refreshAuthenticationCredentialsAndCloseIfTooExpired), service.getConfiguration().getAuthenticationRefreshCheckSeconds(), service.getConfiguration().getAuthenticationRefreshCheckSeconds(), TimeUnit.SECONDS); From e1534ed2034d5cda78288ec6b8dfc2eb4dbcd69a Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Tue, 11 Apr 2023 11:48:53 -0500 Subject: [PATCH 09/12] Fix incorrectly inverted conditional --- .../java/org/apache/pulsar/proxy/server/ProxyConnection.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 59eb3b87d7db5..86c525839afae 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -528,7 +528,7 @@ private void maybeSendAuthChallenge() { LOG.warn("[{}] Closing connection because client doesn't support auth credentials refresh", remoteAddress); ctx.close(); return; - } else if (authChallengeSentTime == Long.MAX_VALUE) { + } else if (authChallengeSentTime != Long.MAX_VALUE) { // If the proxy sent a refresh but hasn't yet heard back, do not send another challenge. return; } From 13ecb1db7369774ae6007713c8377eca1668d35e Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Tue, 11 Apr 2023 12:00:37 -0500 Subject: [PATCH 10/12] Add missed updates to clientAuthData --- .../pulsar/proxy/server/ProxyConnection.java | 17 ++++++++++------- .../proxy/server/ProxyRefreshAuthTest.java | 4 +++- 2 files changed, 13 insertions(+), 8 deletions(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 86c525839afae..a7fc754379147 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -647,6 +647,7 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) { } try { + // Reset the auth challenge sent time to indicate we are not waiting on a client response. authChallengeSentTime = Long.MAX_VALUE; AuthData clientData = AuthData.of(authResponse.getResponse().getAuthData()); // Authenticate the client's auth data and send to the broker concurrently @@ -654,15 +655,17 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) { // from working when forwardAuthorizationCredentials is enabled. Here is an issue to fix the protocol: // https://github.com/apache/pulsar/issues/19291. doAuthentication(clientData); - // We only have pendingBrokerAuthChallenges when forwardAuthorizationCredentials is enabled. - if (pendingBrokerAuthChallenges != null && !pendingBrokerAuthChallenges.isEmpty()) { - // Must store the clientAuthData to be able to initialize future ProxyClientCnx. + if (service.getConfiguration().isForwardAuthorizationCredentials()) { + // Update the clientAuthData to be able to initialize future ProxyClientCnx. this.clientAuthData = clientData; - // Send pending auth data requests to the broker - for (CompletableFuture challenge : pendingBrokerAuthChallenges) { - challenge.complete(clientData); + // We only have pendingBrokerAuthChallenges when forwardAuthorizationCredentials is enabled. + if (pendingBrokerAuthChallenges != null && !pendingBrokerAuthChallenges.isEmpty()) { + // Send auth data to pending challenges from the broker + for (CompletableFuture challenge : pendingBrokerAuthChallenges) { + challenge.complete(clientData); + } + pendingBrokerAuthChallenges.clear(); } - pendingBrokerAuthChallenges.clear(); } } catch (Exception e) { String errorMsg = "Unable to handleAuthResponse"; diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java index d442e32c2f2f8..bde989fc432f9 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyRefreshAuthTest.java @@ -78,7 +78,9 @@ protected void doInitConf() throws Exception { conf.setAuthenticationProviders(Set.of(AuthenticationProviderToken.class.getName())); Properties properties = new Properties(); properties.setProperty("tokenSecretKey", AuthTokenUtils.encodeKeyBase64(SECRET_KEY)); - properties.setProperty("tokenAllowedClockSkewSeconds", "3"); + // The skew should be double the proxy's refresh interval to ensure the broker accepts auth data + // that the proxy might forward. + properties.setProperty("tokenAllowedClockSkewSeconds", "2"); conf.setProperties(properties); conf.setClusterName("proxy-authorization"); From 5b89b453cb8c8ae00623607c49f897db6566dd9f Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Tue, 11 Apr 2023 12:13:47 -0500 Subject: [PATCH 11/12] Minor cleanup --- .../java/org/apache/pulsar/proxy/server/ProxyConnection.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index a7fc754379147..86e908dd65633 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -537,8 +537,8 @@ private void maybeSendAuthChallenge() { LOG.debug("[{}] Refreshing authentication credentials", remoteAddress); } try { - AuthData brokerData = authState.refreshAuthentication(); - writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, getRemoteEndpointProtocolVersion())); + AuthData challenge = authState.refreshAuthentication(); + writeAndFlush(Commands.newAuthChallenge(authMethod, challenge, protocolVersionToAdvertise)); if (LOG.isDebugEnabled()) { LOG.debug("[{}] Sent auth challenge to client to refresh credentials with method: {}.", remoteAddress, authMethod); From ed37db6406d1d7841d317926611ba346fd5f8dd1 Mon Sep 17 00:00:00 2001 From: Michael Marshall Date: Tue, 11 Apr 2023 12:19:41 -0500 Subject: [PATCH 12/12] Prevent leak in ProxyConnection when auth refresh is disabled --- .../org/apache/pulsar/proxy/server/ProxyConnection.java | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 86e908dd65633..2220803a45a6b 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -531,6 +531,13 @@ private void maybeSendAuthChallenge() { } else if (authChallengeSentTime != Long.MAX_VALUE) { // If the proxy sent a refresh but hasn't yet heard back, do not send another challenge. return; + } else if (service.getConfiguration().getAuthenticationRefreshCheckSeconds() < 1) { + // Without the refresh check enabled, there is no way to guarantee the ProxyConnection will close + // this connection if the client fails to respond to the auth challenge with valid auth data. + // The cost is minimal since the client can recreate the connection. This logic prevents a leak. + LOG.warn("[{}] Closing connection because auth credentials refresh is disabled", remoteAddress); + ctx.close(); + return; } if (LOG.isDebugEnabled()) {