[fix][proxy] Refresh auth data if ProxyLookupRequests

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;

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.
  * <p>
  * 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,21 @@ protected void handleAuthChallenge(CommandAuthChallenge authChallenge) {
         checkArgument(authChallenge.getChallenge().hasAuthData());
 
         boolean isRefresh = Arrays.equals(AuthData.REFRESH_AUTH_DATA_BYTES, authChallenge.getChallenge().getAuthData());
-        if (!forwardClientAuthData || !isRefresh) {
-            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 (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())
+                    .exceptionally(ex -> {
+                        log.warn("Failed to get valid client auth data. Closing connection.", ex);
+                        ctx.close();
+                        return null;
                     });
-
-            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);
+        } else {
+            super.handleAuthChallenge(authChallenge);
         }
     }
 }

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."