[client] Restrict the use of client.security.sasl.jaas.config to PlainLoginModule exclusively. (#3425)

Author: loserwang1024

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -1413,7 +1413,10 @@ public class ConfigOptions {
                     .stringType()
                     .noDefaultValue()
                     .withDescription(
-                            "JAAS configuration string for the client. If not provided, uses the JVM option -Djava.security.auth.login.config. \n"
+                            "JAAS configuration string for the client. This option is retained for backward "
+                                    + "compatibility only. Since only SASL/PLAIN is currently supported, only "
+                                    + "PlainLoginModule is accepted. Prefer using 'client.security.sasl.username' "
+                                    + "and 'client.security.sasl.password' directly.\n"
                                     + "Example: org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required username=\"admin\" password=\"admin-secret\";");
 
     public static final ConfigOption<String> CLIENT_SASL_JAAS_USERNAME =

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslClientAuthenticator.java

@@ -20,11 +20,14 @@
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.AuthenticationException;
 import org.apache.fluss.security.auth.ClientAuthenticator;
+import org.apache.fluss.security.auth.sasl.jaas.JaasConfig;
 import org.apache.fluss.security.auth.sasl.jaas.JaasContext;
 import org.apache.fluss.security.auth.sasl.jaas.LoginManager;
+import org.apache.fluss.security.auth.sasl.plain.PlainLoginModule;
 import org.apache.fluss.security.auth.sasl.plain.PlainSaslServer;
 
 import javax.annotation.Nullable;
+import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.auth.login.LoginException;
 import javax.security.sasl.SaslClient;
 
@@ -50,7 +53,13 @@ public class SaslClientAuthenticator implements ClientAuthenticator {
     public SaslClientAuthenticator(Configuration configuration) {
         this.mechanism = configuration.get(CLIENT_SASL_MECHANISM).toUpperCase();
         String jaasConfigStr = configuration.getString(CLIENT_SASL_JAAS_CONFIG);
-        if (jaasConfigStr == null && mechanism.equals(PlainSaslServer.PLAIN_MECHANISM)) {
+        if (jaasConfigStr != null) {
+            // Validate that only PlainLoginModule is allowed in the JAAS config.
+            // Fluss uses a plugin-based authentication system and does not support
+            // custom SASL mechanisms. The jaas.config option is retained for backward
+            // compatibility only.
+            validatePlainLoginModule(jaasConfigStr);
+        } else if (mechanism.equals(PlainSaslServer.PLAIN_MECHANISM)) {
             String username = configuration.get(CLIENT_SASL_JAAS_USERNAME);
             String password = configuration.get(CLIENT_SASL_JAAS_PASSWORD);
             if (username != null || password != null) {
@@ -68,6 +77,39 @@ public SaslClientAuthenticator(Configuration configuration) {
         this.pros = configuration.toMap();
     }
 
+    /**
+     * Validates that the provided JAAS configuration string only uses {@link PlainLoginModule}.
+     *
+     * @param jaasConfigStr the JAAS configuration string to validate
+     * @throws AuthenticationException if the JAAS config uses a login module other than
+     *     PlainLoginModule
+     */
+    private static void validatePlainLoginModule(String jaasConfigStr) {
+        JaasConfig jaasConfig = new JaasConfig("FlussClient", jaasConfigStr);
+        AppConfigurationEntry[] entries = jaasConfig.getAppConfigurationEntry("FlussClient");
+        if (entries == null || entries.length == 0) {
+            throw new AuthenticationException(
+                    "JAAS config property does not contain any login modules");
+        }
+        if (entries.length != 1) {
+            throw new AuthenticationException(
+                    "JAAS config property contains "
+                            + entries.length
+                            + " login modules, should be 1 module");
+        }
+        String loginModuleName = entries[0].getLoginModuleName();
+        if (!PlainLoginModule.class.getName().equals(loginModuleName)) {
+            throw new AuthenticationException(
+                    String.format(
+                            "Only '%s' is supported in '%s'. "
+                                    + "Fluss uses a plugin-based authentication system and does not support "
+                                    + "custom SASL mechanisms. Got: '%s'",
+                            PlainLoginModule.class.getName(),
+                            CLIENT_SASL_JAAS_CONFIG.key(),
+                            loginModuleName));
+        }
+    }
+
     @Override
     public String protocol() {
         return mechanism;

fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java

@@ -78,6 +78,25 @@ void testNormalAuthenticate() throws Exception {
         }
     }
 
+    @Test
+    void testJaasConfigRejectsNonPlainLoginModule() throws Exception {
+        Configuration clientConfig = new Configuration();
+        clientConfig.set(ConfigOptions.CLIENT_SECURITY_PROTOCOL, "sasl");
+        clientConfig.set(ConfigOptions.CLIENT_SASL_MECHANISM, "plain");
+        clientConfig.setString(
+                "client.security.sasl.jaas.config",
+                "com.sun.security.auth.module.JndiLoginModule required username=\"root\" password=\"password\";");
+        try (NettyClient nettyClient =
+                new NettyClient(clientConfig, TestingClientMetricGroup.newInstance())) {
+            assertThatThrownBy(
+                            () -> verifyGetTableNamesList(nettyClient, usernamePasswordServerNode))
+                    .isInstanceOf(AuthenticationException.class)
+                    .hasMessageContaining(
+                            "Only 'org.apache.fluss.security.auth.sasl.plain.PlainLoginModule' is supported")
+                    .hasMessageContaining("Got: 'com.sun.security.auth.module.JndiLoginModule'");
+        }
+    }
+
     @Test
     void testMutualAuthenticate() throws Exception {
         Configuration clientConfig = new Configuration();

fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java

@@ -106,9 +106,9 @@ void testClientLackLoginModule() {
         clientConfig.setString("client.security.sasl.mechanism", "FAKE");
         clientConfig.setString("client.security.sasl.jaas.config", jaasClientInfo);
         assertThatThrownBy(() -> testAuthentication(clientConfig))
-                .cause()
                 .isExactlyInstanceOf(AuthenticationException.class)
-                .hasMessage("Failed to load login manager");
+                .hasMessageContaining(
+                        "Only 'org.apache.fluss.security.auth.sasl.plain.PlainLoginModule' is supported in 'client.security.sasl.jaas.config'.");
     }
 
     @Test
@@ -120,9 +120,9 @@ void testClientMechanismNotMatchServer() {
         clientConfig.setString("client.security.sasl.mechanism", "DIGEST-MD5");
         clientConfig.setString("client.security.sasl.jaas.config", jaasClientInfo);
         assertThatThrownBy(() -> testAuthentication(clientConfig))
-                .cause()
                 .isExactlyInstanceOf(AuthenticationException.class)
-                .hasMessage("SASL server enables [PLAIN] while protocol of client is 'DIGEST-MD5'");
+                .hasMessageContaining(
+                        "Only 'org.apache.fluss.security.auth.sasl.plain.PlainLoginModule' is supported in 'client.security.sasl.jaas.config'.");
     }
 
     @Test