diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java index 25ae1894148..02cf3f33f1d 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java +++ b/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 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 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,6 +154,7 @@ private void enrichWithJaasConfig(Configuration newConfig) throws ConfigExceptio if (Objects.equals(newCredentials, currentPlainCredentials)) { return; } + validateConfigEntries(newCredentials); conf.setString( ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG, @@ -167,6 +162,23 @@ private void enrichWithJaasConfig(Configuration newConfig) throws ConfigExceptio 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 entries) throws ConfigException { + if (entries == null || entries.isEmpty()) { + return; + } + int index = 0; + for (Map.Entry entry : entries.entrySet()) { + validateUsername(entry.getKey()); + validatePassword(index, entry.getKey(), entry.getValue()); + index++; + } + } + private static Map 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)); } } diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java index b8a2a155b06..f3756e34c39 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java +++ b/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");