[client] Simply username and password config for client. (#1072)

Author: loserwang1024

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

@@ -40,7 +40,6 @@
 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.
@@ -1096,17 +1095,35 @@ public class ConfigOptions {
     public static final ConfigOption<String> CLIENT_MECHANISM =
             key("client.security.sasl.mechanism")
                     .stringType()
-                    .noDefaultValue()
+                    .defaultValue("PLAIN")
                     .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)
+            key("client.security.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\")");
+                                    + "Example: com.alibaba.fluss.security.auth.sasl.plain.PlainLoginModule required username=\"admin\" password=\"admin-secret\";");
+
+    public static final ConfigOption<String> CLIENT_SASL_JAAS_USERNAME =
+            key("client.security.sasl.username")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The password to use for client-side SASL JAAS authentication. "
+                                    + "This is used when the client connects to the Fluss cluster with SASL authentication enabled. "
+                                    + "If not provided, the username will be read from the JAAS configuration string specified by `client.security.sasl.jaas.config`.");
+
+    public static final ConfigOption<String> CLIENT_SASL_JAAS_PASSWORD =
+            key("client.security.sasl.password")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The username to use for client-side SASL JAAS authentication. "
+                                    + "This is used when the client connects to the Fluss cluster with SASL authentication enabled. "
+                                    + "If not provided, the password will be read from the JAAS configuration string specified by `client.security.sasl.jaas.config`.");
 
     // ------------------------------------------------------------------------
     //  ConfigOptions for Fluss Table

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

@@ -21,6 +21,7 @@
 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 com.alibaba.fluss.security.auth.sasl.plain.PlainSaslServer;
 
 import javax.annotation.Nullable;
 import javax.security.auth.login.LoginException;
@@ -30,10 +31,14 @@
 
 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.config.ConfigOptions.CLIENT_SASL_JAAS_PASSWORD;
+import static com.alibaba.fluss.config.ConfigOptions.CLIENT_SASL_JAAS_USERNAME;
 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 static final String JAAS_CONF_FORMAT =
+            "com.alibaba.fluss.security.auth.sasl.plain.PlainLoginModule required username=\"%s\" password=\"%s\";";
     private final String mechanism;
     private final Map<String, String> pros;
     private final String jaasConfig;
@@ -43,7 +48,22 @@ public class SaslClientAuthenticator implements ClientAuthenticator {
 
     public SaslClientAuthenticator(Configuration configuration) {
         this.mechanism = configuration.get(CLIENT_MECHANISM).toUpperCase();
-        this.jaasConfig = configuration.getString(CLIENT_SASL_JAAS_CONFIG);
+        String jaasConfigStr = configuration.getString(CLIENT_SASL_JAAS_CONFIG);
+        if (jaasConfigStr == null && 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) {
+                if (username == null || password == null) {
+                    throw new AuthenticationException(
+                            String.format(
+                                    "Configuration '%s' and '%s' must be set together for SASL JAAS authentication",
+                                    CLIENT_SASL_JAAS_USERNAME.key(),
+                                    CLIENT_SASL_JAAS_PASSWORD.key()));
+                }
+                jaasConfigStr = String.format(JAAS_CONF_FORMAT, username, password);
+            }
+        }
+        this.jaasConfig = jaasConfigStr;
         this.pros = configuration.toMap();
     }
 

fluss-rpc/src/test/java/com/alibaba/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java

@@ -161,6 +161,24 @@ void testLoadJassConfigFallBackToJvmOptions() throws Exception {
         testAuthentication(clientConfig, serverConfig);
     }
 
+    @Test
+    void testSimplifyUsernameAndPassword() throws Exception {
+        Configuration clientConfig = new Configuration();
+        clientConfig.setString("client.security.protocol", "sasl");
+        clientConfig.setString("client.security.sasl.username", "alice");
+        assertThatThrownBy(() -> testAuthentication(clientConfig))
+                .isExactlyInstanceOf(AuthenticationException.class)
+                .hasMessage(
+                        "Configuration 'client.security.sasl.username' and 'client.security.sasl.password' must be set together for SASL JAAS authentication");
+        clientConfig.setString("client.security.sasl.password", "wrong-secret");
+        assertThatThrownBy(() -> testAuthentication(clientConfig))
+                .cause()
+                .isExactlyInstanceOf(AuthenticationException.class)
+                .hasMessage("Authentication failed: Invalid username or password");
+        clientConfig.setString("client.security.sasl.password", "alice-secret");
+        testAuthentication(clientConfig);
+    }
+
     private void testAuthentication(Configuration clientConfig) throws Exception {
         testAuthentication(clientConfig, getDefaultServerConfig());
     }