[server][auth] Support SASL/PLAIN authentication. (#985)

Author: loserwang1024

LICENSE

@@ -312,6 +312,19 @@ Apache Kafka
 ./fluss-common/src/main/java/com/alibaba/fluss/utils/log/ByteBufferUnmapper.java
 ./fluss-common/src/main/java/com/alibaba/fluss/utils/log/FairBucketStatusMap.java
 ./fluss-common/src/main/java/com/alibaba/fluss/utils/ExponentialBackoff.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/jaas/AuthenticateCallbackHandler.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/jaas/DefaultLogin.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/jaas/JaasConfig.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/jaas/JaasContext.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/jaas/Login.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/jaas/LoginManager.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/jaas/SaslClientCallbackHandler.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/jaas/SaslServerFactory.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/plain/PlainAuthenticateCallback.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/plain/PlainLoginModule.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/plain/PlainSaslServer.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/plain/PlainSaslServerProvider.java
+./fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/plain/PlainServerCallbackHandler.java
 ./fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/statemachine/ReplicaStateMachine.java
 ./fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/statemachine/TableBucketStateMachine.java
 ./fluss-server/src/main/java/com/alibaba/fluss/server/log/AbstractIndex.java

fluss-common/src/main/java/com/alibaba/fluss/config/ConfigOptions.java

@@ -40,6 +40,7 @@
 import static com.alibaba.fluss.config.ConfigOptions.InfoLogLevel.INFO_LEVEL;
 import static com.alibaba.fluss.config.ConfigOptions.NoKeyAssigner.ROUND_ROBIN;
 import static com.alibaba.fluss.config.ConfigOptions.NoKeyAssigner.STICKY;
+import static com.alibaba.fluss.security.auth.sasl.jaas.JaasContext.SASL_JAAS_CONFIG;
 
 /**
  * Config options for Fluss.
@@ -177,6 +178,9 @@ public class ConfigOptions {
                             "The maximum number of buckets that can be created for a table."
                                     + "The default value is 128000");
 
+    public static final ConfigOption<List<String>> SERVER_SASL_ENABLED_MECHANISMS_CONFIG =
+            key("security.sasl.enabled.mechanisms").stringType().asList().noDefaultValue();
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for Coordinator Server
     // ------------------------------------------------------------------------
@@ -1089,6 +1093,21 @@ public class ConfigOptions {
                             "Enable metrics for client. When metrics is enabled, the client "
                                     + "will collect metrics and report by the JMX metrics reporter.");
 
+    public static final ConfigOption<String> CLIENT_MECHANISM =
+            key("client.security.sasl.mechanism")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "SASL mechanism to use for authentication.Currently, we only support plain.");
+
+    public static final ConfigOption<String> CLIENT_SASL_JAAS_CONFIG =
+            key("client.security.sasl." + SASL_JAAS_CONFIG)
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "JAAS configuration string for the client. If not provided, uses the JVM option -Djava.security.auth.login.config. \n"
+                                    + "Example: com.alibaba.fluss.security.auth.sasl.plain.PlainLoginModule required username=\"admin\" password=\"admin-secret\")");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for Fluss Table
     // ------------------------------------------------------------------------

fluss-common/src/main/java/com/alibaba/fluss/security/auth/AuthenticationFactory.java

@@ -56,8 +56,6 @@
  */
 @PublicEvolving
 public class AuthenticationFactory {
-    private static final String SERVER_AUTHENTICATOR_PREFIX = "security.";
-
     /**
      * Loads a supplier for a client authenticator based on the configuration.
      *

fluss-common/src/main/java/com/alibaba/fluss/security/auth/ClientAuthenticator.java

@@ -21,15 +21,34 @@
 
 import javax.annotation.Nullable;
 
+import java.io.Closeable;
+
 /** Authenticator for client side. */
 @PublicEvolving
-public interface ClientAuthenticator {
+public interface ClientAuthenticator extends Closeable {
 
     /** The protocol name of the authenticator, which will send in the AuthenticateRequest. */
     String protocol();
 
     /** Initialize the authenticator. */
-    default void initialize(AuthenticateContext context) {}
+    default void initialize(AuthenticateContext context) throws AuthenticationException {}
+
+    /**
+     * Determines whether the client authenticator should proactively send an initial token to the
+     * server.
+     *
+     * <p>When this method returns {@code true}, it indicates that the client is the initiator of
+     * the authentication exchange and should actively call {@link #authenticate(byte[])
+     * authenticate(new byte[0])} to generate and send the initial token without waiting for a
+     * challenge from the server.
+     *
+     * @return {@code true} if the client should initiate authentication by sending an initial
+     *     token; {@code false} if the client expects to receive the first token or challenge from
+     *     the server.
+     */
+    default boolean hasInitialTokenResponse() {
+        return true;
+    }
 
     /**
      * * Generates the initial token or calculates a token based on the server's challenge, then
@@ -79,6 +98,10 @@ default void initialize(AuthenticateContext context) {}
     /** Checks if the authentication from client side is completed. */
     boolean isCompleted();
 
+    default void close() {}
+
     /** The context of the authentication process. */
-    interface AuthenticateContext {}
+    interface AuthenticateContext {
+        String ipAddress();
+    }
 }

fluss-common/src/main/java/com/alibaba/fluss/security/auth/ServerAuthenticator.java

@@ -20,16 +20,28 @@
 import com.alibaba.fluss.exception.AuthenticationException;
 import com.alibaba.fluss.security.acl.FlussPrincipal;
 
+import java.io.Closeable;
+import java.io.IOException;
+
 /**
  * Authenticator for server side.
  *
  * @since 0.7
  */
 @PublicEvolving
-public interface ServerAuthenticator {
+public interface ServerAuthenticator extends Closeable {
 
     String protocol();
 
+    default void matchProtocol(String protocol) throws AuthenticationException {
+        if (!protocol().equalsIgnoreCase(protocol)) {
+            throw new AuthenticationException(
+                    String.format(
+                            "Authenticate protocol not match: protocol of server is '%s' while protocol of client is '%s'",
+                            protocol(), protocol));
+        }
+    }
+
     /** Initialize the authenticator. */
     default void initialize(AuthenticateContext context) {}
 
@@ -83,6 +95,15 @@ default void initialize(AuthenticateContext context) {}
      */
     FlussPrincipal createPrincipal();
 
+    /** Close the authenticator. */
+    default void close() throws IOException {}
+
     /** The context of the authentication process. */
-    interface AuthenticateContext {}
+    interface AuthenticateContext {
+        String ipAddress();
+
+        String listenerName();
+
+        String protocol();
+    }
 }

fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/authenticator/SaslAuthenticationPlugin.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.security.auth.sasl.authenticator;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.security.auth.ClientAuthenticationPlugin;
+import com.alibaba.fluss.security.auth.ClientAuthenticator;
+import com.alibaba.fluss.security.auth.ServerAuthenticationPlugin;
+import com.alibaba.fluss.security.auth.ServerAuthenticator;
+
+/** Authentication plugin for SASL. */
+public class SaslAuthenticationPlugin
+        implements ClientAuthenticationPlugin, ServerAuthenticationPlugin {
+    static final String SASL_AUTH_PROTOCOL = "sasl";
+
+    @Override
+    public ClientAuthenticator createClientAuthenticator(Configuration configuration) {
+        return new SaslClientAuthenticator(configuration);
+    }
+
+    @Override
+    public ServerAuthenticator createServerAuthenticator(Configuration configuration) {
+        return new SaslServerAuthenticator(configuration);
+    }
+
+    @Override
+    public String authProtocol() {
+        return SASL_AUTH_PROTOCOL;
+    }
+}

fluss-common/src/main/java/com/alibaba/fluss/security/auth/sasl/authenticator/SaslClientAuthenticator.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.security.auth.sasl.authenticator;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.exception.AuthenticationException;
+import com.alibaba.fluss.security.auth.ClientAuthenticator;
+import com.alibaba.fluss.security.auth.sasl.jaas.JaasContext;
+import com.alibaba.fluss.security.auth.sasl.jaas.LoginManager;
+
+import javax.annotation.Nullable;
+import javax.security.auth.login.LoginException;
+import javax.security.sasl.SaslClient;
+
+import java.util.Map;
+
+import static com.alibaba.fluss.config.ConfigOptions.CLIENT_MECHANISM;
+import static com.alibaba.fluss.config.ConfigOptions.CLIENT_SASL_JAAS_CONFIG;
+import static com.alibaba.fluss.security.auth.sasl.jaas.SaslServerFactory.createSaslClient;
+
+/** An authenticator that uses SASL to authenticate with a server. */
+public class SaslClientAuthenticator implements ClientAuthenticator {
+    private final String mechanism;
+    private final Map<String, String> pros;
+    private final String jaasConfig;
+
+    private SaslClient saslClient;
+    private LoginManager loginManager;
+
+    public SaslClientAuthenticator(Configuration configuration) {
+        this.mechanism = configuration.get(CLIENT_MECHANISM).toUpperCase();
+        this.jaasConfig = configuration.getString(CLIENT_SASL_JAAS_CONFIG);
+        this.pros = configuration.toMap();
+    }
+
+    @Override
+    public String protocol() {
+        return mechanism;
+    }
+
+    @Nullable
+    @Override
+    public byte[] authenticate(byte[] data) throws AuthenticationException {
+        try {
+            return saslClient.evaluateChallenge(data);
+        } catch (Exception e) {
+            throw new AuthenticationException("Failed to evaluate SASL challenge", e);
+        }
+    }
+
+    @Override
+    public boolean isCompleted() {
+        return saslClient.isComplete();
+    }
+
+    @Override
+    public boolean hasInitialTokenResponse() {
+        return saslClient.hasInitialResponse();
+    }
+
+    @Override
+    public void initialize(AuthenticateContext context) throws AuthenticationException {
+        String hostAddress = context.ipAddress();
+        JaasContext jaasContext = JaasContext.loadClientContext(jaasConfig);
+
+        try {
+            loginManager = LoginManager.acquireLoginManager(jaasContext);
+        } catch (LoginException exception) {
+            throw new AuthenticationException("Failed to load login manager", exception);
+        }
+
+        try {
+            saslClient = createSaslClient(mechanism, hostAddress, pros, loginManager);
+        } catch (Exception e) {
+            throw new AuthenticationException("Failed to create SASL client", e);
+        }
+
+        if (saslClient == null) {
+            throw new AuthenticationException(
+                    "Unable to find a matching SASL mechanism for " + mechanism);
+        }
+    }
+
+    @Override
+    public void close() {
+        if (loginManager != null) {
+            loginManager.release();
+        }
+    }
+}