[rpc] Validate config entries consistently in FlussProtocolPlugin

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java

@@ -53,13 +53,14 @@ public class FlussProtocolPlugin implements NetworkProtocolPlugin, ServerReconfi
     private static final Pattern VALID_USERNAME_PATTERN = Pattern.compile("\\w+");
 
     /**
-     * Characters forbidden in passwords. These would break the map format or the generated JAAS
-     * config string: comma (entry separator), colon (key-value separator), double-quote (JAAS value
-     * delimiter), semicolon (JAAS statement terminator), backslash (escape char), and control
-     * characters.
+     * Characters that cannot be represented in the generated config value and are therefore
+     * rejected on both the setup and reconfigure paths: the double-quote and backslash characters,
+     * which carry special meaning in the generated value, and control characters. Separators such
+     * as commas, colons and semicolons are handled by the map parser and, once quoted, may
+     * legitimately appear in a value, so they are accepted here.
      */
     private static final Pattern INVALID_PASSWORD_PATTERN =
-            Pattern.compile("[,:\"\\\\;]|[\\x00-\\x1F\\x7F]");
+            Pattern.compile("[\"\\\\]|[\\x00-\\x1F\\x7F]");
 
     private final ApiManager apiManager;
     private final List<String> listeners;
@@ -125,14 +126,7 @@ public void validate(Configuration newConfig) throws ConfigException {
         if (Objects.equals(newCredentials, currentPlainCredentials)) {
             return;
         }
-        if (newCredentials != null && !newCredentials.isEmpty()) {
-            int index = 0;
-            for (Map.Entry<String, String> credential : newCredentials.entrySet()) {
-                validateUsername(credential.getKey());
-                validatePassword(index, credential.getKey(), credential.getValue());
-                index++;
-            }
-        }
+        validateConfigEntries(newCredentials);
 
         // Generate the merged JAAS config value to ensure it is valid.
         generateMergedJaasConfig(newCredentials);
@@ -160,13 +154,31 @@ private void enrichWithJaasConfig(Configuration newConfig) throws ConfigExceptio
         if (Objects.equals(newCredentials, currentPlainCredentials)) {
             return;
         }
+        validateConfigEntries(newCredentials);
 
         conf.setString(
                 ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG,
                 generateMergedJaasConfig(newCredentials));
         currentPlainCredentials = newCredentials;
     }
 
+    /**
+     * Validates each configuration entry so that the same rules apply on the setup path and the
+     * reconfigure path. Entries whose values cannot be represented in the generated config are
+     * rejected here rather than silently corrupting it.
+     */
+    private static void validateConfigEntries(Map<String, String> entries) throws ConfigException {
+        if (entries == null || entries.isEmpty()) {
+            return;
+        }
+        int index = 0;
+        for (Map.Entry<String, String> entry : entries.entrySet()) {
+            validateUsername(entry.getKey());
+            validatePassword(index, entry.getKey(), entry.getValue());
+            index++;
+        }
+    }
+
     private static Map<String, String> readPlainCredentials(Configuration config)
             throws ConfigException {
         try {
@@ -195,7 +207,7 @@ private static void validatePassword(int index, String username, String password
             throw new ConfigException(
                     String.format(
                             "%s[%d]: password for user '%s' contains invalid characters. "
-                                    + "Commas, colons, quotes, semicolons, backslashes, and control characters are not allowed.",
+                                    + "Double-quote, backslash, and control characters are not allowed.",
                             PLAIN_CREDENTIALS_CONFIG, index, username));
         }
     }

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

@@ -439,26 +439,22 @@ void testValidateRejectsInvalidPasswordCharacters() throws Exception {
                     .isInstanceOf(ConfigException.class)
                     .hasMessageContaining("Failed to parse security.sasl.plain.credentials");
 
-            // Password with colon (breaks map key-value format)
+            // A colon inside a quoted value is representable and should be accepted
             Configuration colonPassword = new Configuration();
             colonPassword.setString("security.sasl.plain.credentials", "bob:'pass:word'");
-            assertThatThrownBy(() -> reconfigurable.validate(colonPassword))
-                    .isInstanceOf(ConfigException.class)
-                    .hasMessageContaining("contains invalid characters");
+            reconfigurable.validate(colonPassword);
 
-            // Password with double-quote (breaks JAAS value)
+            // A double-quote cannot be represented in the generated value
             Configuration quotePassword = new Configuration();
             quotePassword.setString("security.sasl.plain.credentials", "bob:pass\"word");
             assertThatThrownBy(() -> reconfigurable.validate(quotePassword))
                     .isInstanceOf(ConfigException.class)
                     .hasMessageContaining("password for user 'bob' contains invalid characters");
 
-            // Password with semicolon (breaks JAAS statement)
+            // A semicolon is representable inside the quoted value and should be accepted
             Configuration semicolonPassword = new Configuration();
             semicolonPassword.setString("security.sasl.plain.credentials", "bob:pass;word");
-            assertThatThrownBy(() -> reconfigurable.validate(semicolonPassword))
-                    .isInstanceOf(ConfigException.class)
-                    .hasMessageContaining("contains invalid characters");
+            reconfigurable.validate(semicolonPassword);
 
             // Password with backslash (escape char)
             Configuration backslashPassword = new Configuration();
@@ -474,6 +470,82 @@ void testValidateRejectsInvalidPasswordCharacters() throws Exception {
         }
     }
 
+    /**
+     * The setup path must apply the same per-entry validation as the reconfigure path: a config
+     * value that cannot be represented in the generated config is rejected loudly at startup, while
+     * a value the map splitter legitimately supports is accepted and produces a working config.
+     */
+    @Test
+    void testSetupRejectsInvalidConfigValueAndAcceptsRepresentableValue() throws Exception {
+        MetricGroup metricGroup = NOPMetricsGroup.newInstance();
+        TestingAuthenticateGatewayService service = new TestingAuthenticateGatewayService();
+
+        // A value containing a double-quote cannot be represented in the generated config and must
+        // be rejected while the server is being set up, before it starts.
+        try (NetUtils.Port port = getAvailablePort()) {
+            Configuration invalidConfig = new Configuration();
+            invalidConfig.setString(
+                    ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(), "CLIENT:sasl");
+            invalidConfig.setString("security.sasl.enabled.mechanisms", "plain");
+            invalidConfig.setString(
+                    "security.sasl.plain.jaas.config",
+                    "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required"
+                            + " user_admin=\"admin-secret\";");
+            invalidConfig.setString("security.sasl.plain.credentials", "bob:pass\"word");
+            invalidConfig.setString(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS.key(), "3");
+
+            assertThatThrownBy(
+                            () ->
+                                    new NettyServer(
+                                            invalidConfig,
+                                            Collections.singletonList(
+                                                    new Endpoint(
+                                                            "localhost", port.getPort(), "CLIENT")),
+                                            service,
+                                            metricGroup,
+                                            RequestsMetrics.createCoordinatorServerRequestMetrics(
+                                                    metricGroup)))
+                    .isInstanceOf(ConfigException.class)
+                    .hasMessageContaining("contains invalid characters");
+        }
+
+        // A quoted colon is representable, so the same setup path must accept it and produce a
+        // config that lets the configured user connect with that exact value.
+        try (NetUtils.Port port = getAvailablePort()) {
+            Configuration validConfig = new Configuration();
+            validConfig.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(), "CLIENT:sasl");
+            validConfig.setString("security.sasl.enabled.mechanisms", "plain");
+            validConfig.setString(
+                    "security.sasl.plain.jaas.config",
+                    "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required"
+                            + " user_admin=\"admin-secret\";");
+            validConfig.setString("security.sasl.plain.credentials", "bob:'pass:word'");
+            validConfig.setString(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS.key(), "3");
+
+            try (NettyServer nettyServer =
+                    new NettyServer(
+                            validConfig,
+                            Collections.singletonList(
+                                    new Endpoint("localhost", port.getPort(), "CLIENT")),
+                            service,
+                            metricGroup,
+                            RequestsMetrics.createCoordinatorServerRequestMetrics(metricGroup))) {
+                nettyServer.start();
+                ServerNode serverNode =
+                        new ServerNode(1, "localhost", port.getPort(), ServerType.TABLET_SERVER);
+
+                // Existing user from the initial config can still connect.
+                try (NettyClient client = createSaslClient("admin", "admin-secret")) {
+                    verifyListTables(client, serverNode);
+                }
+                // The configured user connects with the exact value that carried a colon.
+                try (NettyClient client = createSaslClient("bob", "pass:word")) {
+                    verifyListTables(client, serverNode);
+                }
+            }
+        }
+    }
+
     private NettyClient createSaslClient(String username, String password) {
         Configuration clientConfig = new Configuration();
         clientConfig.setString("client.security.protocol", "sasl");