You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2026/04/09 10:06:05 UTC
[PR] [server] Manager users for sasl/plain authentication via cluster properties. [fluss]
loserwang1024 opened a new pull request, #3047:
URL: https://github.com/apache/fluss/pull/3047
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #3046
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Manage users for sasl/plain authentication via cluster properties. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #3047:
URL: https://github.com/apache/fluss/pull/3047#issuecomment-4900773035
@wuchong ， I have modified this PR based on your advice. Only one is left for another PR later.
One concern I have is that `security.sasl.plain.credentials` is a bit different from normal sensitive config options. Each list entry stores `username:password` as a single string value, so the config framework cannot naturally distinguish the non-sensitive username part from the sensitive password part.
If we only add a generic sensitive flag, the safest behavior would be to mask the whole config value in `describeClusterConfigs` / `get_cluster_configs`. That is reusable and avoids leaking passwords, but it also means usernames are hidden together with passwords.
If we want to show usernames while hiding passwords, we would need special parsing logic for this specific config key, e.g. rendering `alice:******,bob:******`. I’m not sure that belongs in the generic config display path, because it would make the config framework aware of SASL/PLAIN-specific value syntax and edge cases.
So I’d like to discuss which direction we prefer:
1. Treat the whole `security.sasl.plain.credentials` value as sensitive and mask it entirely.
2. Add special handling for this key to expose usernames but mask passwords.
3. Keep config display generic, and if listing SASL/PLAIN users is needed, add a dedicated user-management/list-users API later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Manage users for sasl/plain authentication via cluster properties. [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #3047:
URL: https://github.com/apache/fluss/pull/3047
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Manage users for sasl/plain authentication via cluster properties. [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3047:
URL: https://github.com/apache/fluss/pull/3047#discussion_r3061773588
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java:
##########
@@ -81,8 +90,73 @@ public RequestHandler<?> createRequestHandler(RpcGatewayService service) {
return new FlussRequestHandler(service);
}
+ // --- ServerReconfigurable ---
+
+ @Override
+ public void validate(Configuration newConfig) throws ConfigException {
+ List<String> users = newConfig.get(ConfigOptions.SERVER_SASL_USERS);
+ if (users == null) {
+ return;
+ }
+ Set<String> uniqueUsernames = new HashSet<>();
+ for (int i = 0; i < users.size(); i++) {
+ String entry = users.get(i).trim();
+ int colonIdx = entry.indexOf(':');
+ if (colonIdx <= 0 || colonIdx == entry.length() - 1) {
+ throw new ConfigException(
+ String.format(
+ "security.sasl.users[%d] must be in 'username:password' format, but got '%s'.",
+ i, entry));
+ }
+ String username = entry.substring(0, colonIdx);
+ if (!uniqueUsernames.add(username)) {
+ throw new ConfigException(
+ "security.sasl.users must not contain duplicate usernames: '"
+ + username
+ + "'.");
+ }
+ }
+ }
Review Comment:
The generated JAAS config later uses `user_%s=\"%s\"` with raw username/password. Without validating/escaping, usernames may contain characters invalid for JAAS option keys (and passwords may contain quotes/backslashes), leading to invalid configs or potential injection into the JAAS string. Recommend: (1) validate username against a strict allowlist suitable for `user_<name>` keys, (2) either disallow or correctly escape `\"` and `\\` in passwords (and other problematic characters), and (3) trim username/password components consistently (currently only the whole entry is trimmed in `validate()`).
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java:
##########
@@ -81,8 +90,73 @@ public RequestHandler<?> createRequestHandler(RpcGatewayService service) {
return new FlussRequestHandler(service);
}
+ // --- ServerReconfigurable ---
+
+ @Override
+ public void validate(Configuration newConfig) throws ConfigException {
+ List<String> users = newConfig.get(ConfigOptions.SERVER_SASL_USERS);
+ if (users == null) {
+ return;
+ }
+ Set<String> uniqueUsernames = new HashSet<>();
+ for (int i = 0; i < users.size(); i++) {
+ String entry = users.get(i).trim();
+ int colonIdx = entry.indexOf(':');
+ if (colonIdx <= 0 || colonIdx == entry.length() - 1) {
+ throw new ConfigException(
+ String.format(
+ "security.sasl.users[%d] must be in 'username:password' format, but got '%s'.",
+ i, entry));
+ }
+ String username = entry.substring(0, colonIdx);
+ if (!uniqueUsernames.add(username)) {
+ throw new ConfigException(
+ "security.sasl.users must not contain duplicate usernames: '"
+ + username
+ + "'.");
+ }
+ }
+ }
+
+ @Override
+ public void reconfigure(Configuration newConfig) throws ConfigException {
+ this.conf = enrichWithJaasConfig(newConfig);
+ }
+
+ /**
+ * Enriches the given configuration with a generated JAAS config string derived from the
+ * security.sasl.users list (format: 'username:password'). If the list is not present, the
+ * original configuration is returned unchanged.
+ */
+ private static Configuration enrichWithJaasConfig(Configuration config) {
+ List<String> users = config.get(ConfigOptions.SERVER_SASL_USERS);
+ if (users == null) {
+ return config;
+ }
+ StringBuilder sb =
+ new StringBuilder(
+ "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required");
+ for (String entry : users) {
+ int colonIdx = entry.indexOf(':');
+ checkArgument(colonIdx > 0, "Invalid user entry format: '%s'", entry);
+ String username = entry.substring(0, colonIdx);
+ String password = entry.substring(colonIdx + 1);
+ sb.append(String.format(" user_%s=\"%s\"", username, password));
+ }
+ sb.append(";");
+ Configuration enriched = new Configuration(config);
+ enriched.setString("security.sasl.plain.jaas.config", sb.toString());
+ return enriched;
+ }
Review Comment:
`enrichWithJaasConfig()` uses `checkArgument(...)` which throws `IllegalArgumentException`, even though the surrounding reconfiguration path is declared to throw `ConfigException` and other code/tests appear to expect config-related failures as `ConfigException`. To keep error handling consistent (and avoid unexpected runtime exceptions), convert this validation to throw `ConfigException` with a user-facing message, and apply the same trimming logic here as in `validate()` (currently `validate()` trims but enrichment does not).
##########
fluss-common/src/main/java/org/apache/fluss/security/auth/AuthenticationFactory.java:
##########
@@ -83,7 +83,8 @@ public static Supplier<ClientAuthenticator> loadClientAuthenticatorSupplier(
* authenticators.
*/
public static Map<String, Supplier<ServerAuthenticator>> loadServerAuthenticatorSuppliers(
Review Comment:
This changes a public static method signature from `(Configuration)` to `(Supplier<Configuration>)`, which is a source/binary breaking API change for any external callers. Consider re-introducing an overload `loadServerAuthenticatorSuppliers(Configuration configuration)` that delegates to the new supplier-based method (optionally deprecate it) to preserve compatibility.
```suggestion
public static Map<String, Supplier<ServerAuthenticator>> loadServerAuthenticatorSuppliers(
Configuration configuration) {
return loadServerAuthenticatorSuppliers(() -> configuration);
}
/**
* Loads suppliers for server authenticators for each endpoint, based on listener-specific
* protocols.
*
* @param configurationSupplier The configuration supplier containing authentication settings
* and protocol definitions.
* @return A map mapping listener names to suppliers for their corresponding server
* authenticators.
*/
public static Map<String, Supplier<ServerAuthenticator>> loadServerAuthenticatorSuppliers(
```
##########
fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java:
##########
@@ -145,6 +148,30 @@ private void prepareIncrementalConfigs(
case DELETE:
configsProps.remove(configPropName);
break;
+ case APPEND:
+ validateListType(configPropName);
+ String existingAppend = configsProps.getOrDefault(configPropName, "");
+ if (existingAppend.isEmpty()) {
+ configsProps.put(configPropName, configPropValue);
+ } else {
+ configsProps.put(
+ configPropName, existingAppend + "," + configPropValue);
+ }
+ break;
+ case SUBTRACT:
+ validateListType(configPropName);
+ String existingSubtract = configsProps.get(configPropName);
+ if (existingSubtract != null) {
+ List<String> items =
+ new ArrayList<>(Arrays.asList(existingSubtract.split(",")));
+ items.remove(configPropValue);
+ if (items.isEmpty()) {
+ configsProps.remove(configPropName);
+ } else {
+ configsProps.put(configPropName, String.join(",", items));
+ }
+ }
+ break;
Review Comment:
SUBTRACT uses a raw `split(\",\")` and then `items.remove(configPropValue)`. This will fail to remove entries if the stored list contains whitespace (e.g. `\"a, b\"` -> item is `\" b\"`), and it only removes the first matching occurrence. Consider normalizing items (trim each element and filter empty segments) and removing all matches (e.g., `removeIf(v -> v.equals(configPropValue))`) so SUBTRACT behaves predictably.
##########
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java:
##########
@@ -301,6 +302,8 @@ protected void startServices() throws Exception {
tabletService,
tabletServerMetricGroup,
requestsMetrics);
+ // Register FlussProtocolPlugin for dynamic SASL config updates
+ dynamicConfigManager.register(((NettyServer) rpcServer).getFlussProtocolPlugin());
Review Comment:
Same issue as in `CoordinatorServer`: the unconditional cast to `NettyServer` is brittle and can crash startup with `ClassCastException`. Use a safer integration point (interface method) or an `instanceof` guard with a clear error explaining why dynamic SASL updates are unavailable/unsupported without NettyServer.
```suggestion
// Register FlussProtocolPlugin for dynamic SASL config updates when supported.
if (rpcServer instanceof NettyServer) {
dynamicConfigManager.register(((NettyServer) rpcServer).getFlussProtocolPlugin());
} else {
LOG.warn(
"Dynamic SASL config updates are unavailable because RpcServer implementation {} "
+ "does not support Netty FlussProtocolPlugin integration.",
rpcServer.getClass().getName());
}
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java:
##########
@@ -783,6 +785,119 @@ void testListRebalanceProgress() throws Exception {
});
}
+ @Test
+ void testAddAndDeleteUser() throws Exception {
+ String bobBootstrapServers =
+ String.join(
+ ",",
+ FLUSS_CLUSTER_EXTENSION
+ .getClientConfig("CLIENT")
+ .get(ConfigOptions.BOOTSTRAP_SERVERS));
+ String bobCatalog = "bob_catalog";
+ String createCatalogDDL =
+ String.format(
+ "create catalog %s with ("
+ + "'type' = 'fluss', "
+ + "'bootstrap.servers' = '%s', "
+ + "'client.security.protocol' = 'sasl', "
+ + "'client.security.sasl.mechanism' = 'PLAIN', "
+ + "'client.security.sasl.username' = 'bob', "
+ + "'client.security.sasl.password' = 'bob_pass'"
+ + ")",
+ bobCatalog, bobBootstrapServers);
+
+ assertThatThrownBy(() -> tEnv.executeSql(createCatalogDDL).await())
+ .hasMessageContaining("Invalid username or password");
+
+ // Step 1: Add user "bob" via append_cluster_configs
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.append_cluster_configs('%s', 'bob:bob_pass')",
+ CATALOG_NAME, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ assertThat(results.get(0).getField(0))
+ .asString()
+ .contains("Successfully appended")
+ .contains(ConfigOptions.SERVER_SASL_USERS.key());
+ }
+
+ // Verify user "bob" was added
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.get_cluster_configs('%s')",
+ CATALOG_NAME, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ assertThat((String) results.get(0).getField(1)).contains("bob:bob_pass");
+ }
+
+ // Verify "bob" can authenticate by creating a catalog with bob's credentials
+ tEnv.executeSql(createCatalogDDL).await();
+
+ // Grant bob DESCRIBE permission on cluster so bob can query configs
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.add_acl('CLUSTER', 'ALLOW', 'User:bob', 'DESCRIBE', '*')",
+ CATALOG_NAME))
+ .await();
+
+ // Bob should be able to get cluster configs
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.get_cluster_configs('%s')",
+ bobCatalog, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ }
+ tEnv.executeSql("drop catalog " + bobCatalog);
Review Comment:
This IT adds cluster state (catalog + ACL) but cleanup is not protected by a `finally` block. If an assertion or intermediate call fails after creating the catalog or adding the ACL, the test can leak state and make subsequent tests flaky. Wrap catalog creation/ACL grant in `try`/`finally` to reliably drop the catalog and remove the ACL.
```suggestion
boolean bobCatalogCreated = false;
boolean bobDescribeAclAdded = false;
try {
// Verify "bob" can authenticate by creating a catalog with bob's credentials
tEnv.executeSql(createCatalogDDL).await();
bobCatalogCreated = true;
// Grant bob DESCRIBE permission on cluster so bob can query configs
tEnv.executeSql(
String.format(
"Call %s.sys.add_acl('CLUSTER', 'ALLOW', 'User:bob', 'DESCRIBE', '*')",
CATALOG_NAME))
.await();
bobDescribeAclAdded = true;
// Bob should be able to get cluster configs
try (CloseableIterator<Row> resultIterator =
tEnv.executeSql(
String.format(
"Call %s.sys.get_cluster_configs('%s')",
bobCatalog, ConfigOptions.SERVER_SASL_USERS.key()))
.collect()) {
List<Row> results = CollectionUtil.iteratorToList(resultIterator);
assertThat(results).hasSize(1);
}
} finally {
try {
if (bobDescribeAclAdded) {
tEnv.executeSql(
String.format(
"Call %s.sys.remove_acl('CLUSTER', 'ALLOW', 'User:bob', 'DESCRIBE', '*')",
CATALOG_NAME))
.await();
}
} finally {
if (bobCatalogCreated) {
tEnv.executeSql("drop catalog " + bobCatalog).await();
}
}
}
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java:
##########
@@ -783,6 +785,119 @@ void testListRebalanceProgress() throws Exception {
});
}
+ @Test
+ void testAddAndDeleteUser() throws Exception {
+ String bobBootstrapServers =
+ String.join(
+ ",",
+ FLUSS_CLUSTER_EXTENSION
+ .getClientConfig("CLIENT")
+ .get(ConfigOptions.BOOTSTRAP_SERVERS));
+ String bobCatalog = "bob_catalog";
+ String createCatalogDDL =
+ String.format(
+ "create catalog %s with ("
+ + "'type' = 'fluss', "
+ + "'bootstrap.servers' = '%s', "
+ + "'client.security.protocol' = 'sasl', "
+ + "'client.security.sasl.mechanism' = 'PLAIN', "
+ + "'client.security.sasl.username' = 'bob', "
+ + "'client.security.sasl.password' = 'bob_pass'"
+ + ")",
+ bobCatalog, bobBootstrapServers);
+
+ assertThatThrownBy(() -> tEnv.executeSql(createCatalogDDL).await())
+ .hasMessageContaining("Invalid username or password");
+
+ // Step 1: Add user "bob" via append_cluster_configs
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.append_cluster_configs('%s', 'bob:bob_pass')",
+ CATALOG_NAME, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ assertThat(results.get(0).getField(0))
+ .asString()
+ .contains("Successfully appended")
+ .contains(ConfigOptions.SERVER_SASL_USERS.key());
+ }
+
+ // Verify user "bob" was added
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.get_cluster_configs('%s')",
+ CATALOG_NAME, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ assertThat((String) results.get(0).getField(1)).contains("bob:bob_pass");
+ }
+
+ // Verify "bob" can authenticate by creating a catalog with bob's credentials
+ tEnv.executeSql(createCatalogDDL).await();
+
+ // Grant bob DESCRIBE permission on cluster so bob can query configs
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.add_acl('CLUSTER', 'ALLOW', 'User:bob', 'DESCRIBE', '*')",
+ CATALOG_NAME))
+ .await();
+
+ // Bob should be able to get cluster configs
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.get_cluster_configs('%s')",
+ bobCatalog, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ }
+ tEnv.executeSql("drop catalog " + bobCatalog);
+
+ // Step 2: Delete user "bob" via subtract_cluster_configs
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.subtract_cluster_configs('%s', 'bob:bob_pass')",
+ CATALOG_NAME, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ assertThat(results.get(0).getField(0))
+ .asString()
+ .contains("Successfully subtracted")
+ .contains(ConfigOptions.SERVER_SASL_USERS.key());
+ }
+
+ // Verify "bob" was deleted from config
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.get_cluster_configs('%s')",
+ CATALOG_NAME, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ // After subtracting the only dynamically-added entry, the config may be empty
+ if (!results.isEmpty()) {
+ assertThat((String) results.get(0).getField(1)).doesNotContain("bob");
+ }
+ }
+
+ // Verify "bob" can no longer authenticate
+ assertThatThrownBy(() -> tEnv.executeSql(createCatalogDDL).await())
+ .hasMessageContaining("Invalid username or password");
+ // Cleanup: remove bob's ACL
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.drop_acl('CLUSTER', 'ALLOW', 'User:bob', 'DESCRIBE', '*')",
+ CATALOG_NAME))
+ .await();
Review Comment:
This IT adds cluster state (catalog + ACL) but cleanup is not protected by a `finally` block. If an assertion or intermediate call fails after creating the catalog or adding the ACL, the test can leak state and make subsequent tests flaky. Wrap catalog creation/ACL grant in `try`/`finally` to reliably drop the catalog and remove the ACL.
```suggestion
try {
try (CloseableIterator<Row> resultIterator =
tEnv.executeSql(
String.format(
"Call %s.sys.subtract_cluster_configs('%s', 'bob:bob_pass')",
CATALOG_NAME, ConfigOptions.SERVER_SASL_USERS.key()))
.collect()) {
List<Row> results = CollectionUtil.iteratorToList(resultIterator);
assertThat(results).hasSize(1);
assertThat(results.get(0).getField(0))
.asString()
.contains("Successfully subtracted")
.contains(ConfigOptions.SERVER_SASL_USERS.key());
}
// Verify "bob" was deleted from config
try (CloseableIterator<Row> resultIterator =
tEnv.executeSql(
String.format(
"Call %s.sys.get_cluster_configs('%s')",
CATALOG_NAME, ConfigOptions.SERVER_SASL_USERS.key()))
.collect()) {
List<Row> results = CollectionUtil.iteratorToList(resultIterator);
// After subtracting the only dynamically-added entry, the config may be empty
if (!results.isEmpty()) {
assertThat((String) results.get(0).getField(1)).doesNotContain("bob");
}
}
// Verify "bob" can no longer authenticate
assertThatThrownBy(() -> tEnv.executeSql(createCatalogDDL).await())
.hasMessageContaining("Invalid username or password");
} finally {
// Cleanup: remove bob's ACL
tEnv.executeSql(
String.format(
"Call %s.sys.drop_acl('CLUSTER', 'ALLOW', 'User:bob', 'DESCRIBE', '*')",
CATALOG_NAME))
.await();
}
```
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java:
##########
@@ -272,6 +272,16 @@ private static NetworkProtocolPlugin loadProtocolPlugin(String protocolName) {
protocolName, protocols.keySet()));
}
+ /** Returns the FlussProtocolPlugin instance used by this server. */
+ public FlussProtocolPlugin getFlussProtocolPlugin() {
+ for (NetworkProtocolPlugin protocol : protocols) {
+ if (protocol instanceof FlussProtocolPlugin) {
+ return (FlussProtocolPlugin) protocol;
+ }
+ }
+ throw new IllegalStateException("FlussProtocolPlugin not found in loaded protocols.");
+ }
Review Comment:
This introduces a hard failure if `FlussProtocolPlugin` is not present in `protocols`. Since callers use this to wire dynamic SASL updates, consider making the contract explicit and easier to handle (e.g., return `Optional<FlussProtocolPlugin>` or provide a boolean/nullable getter) so startup can either (a) clearly error with context about why the plugin is required, or (b) skip registration when the protocol isn't loaded.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java:
##########
@@ -285,6 +286,8 @@ protected void initCoordinatorStandby() throws Exception {
serverMetricGroup,
RequestsMetrics.createCoordinatorServerRequestMetrics(
serverMetricGroup));
+ // Register FlussProtocolPlugin for dynamic SASL config updates
Review Comment:
This hard-casts `rpcServer` to `NettyServer`. If `RpcServer` can be implemented by other server types (now or in the future), this will throw `ClassCastException` during startup. Prefer either (a) exposing the needed hook via the `RpcServer` abstraction, or (b) guarding with `instanceof NettyServer` and failing with a clear `ConfigException/IllegalStateException` that explains the requirement.
```suggestion
// Register FlussProtocolPlugin for dynamic SASL config updates
if (!(rpcServer instanceof NettyServer)) {
throw new IllegalStateException(
"CoordinatorServer requires RpcServer to be a NettyServer for dynamic "
+ "SASL config updates, but got: "
+ rpcServer.getClass().getName());
}
```
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -476,6 +476,16 @@ public class ConfigOptions {
+ "Each listener can be associated with a specific authentication protocol. "
+ "Listeners not included in the map will use PLAINTEXT by default, which does not require authentication.");
+ public static final ConfigOption<List<String>> SERVER_SASL_USERS =
+ key("security.sasl.users")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ "List of user credentials for SASL/PLAIN authentication in 'username:password' format. "
+ + "For example: 'admin:admin-secret,bob:bob-secret'. "
+ + "This is syntactic sugar that auto-generates the JAAS config string.");
Review Comment:
This option stores plaintext credentials and is now exposed as a regular cluster config. To reduce accidental leakage (logs, `get_cluster_configs`, UIs), consider marking this option as sensitive (if the config framework supports it) and/or ensuring config listing APIs mask/redact values for `security.sasl.users` (and the derived `security.sasl.plain.jaas.config`). Also consider documenting the security implications explicitly in the description.
```suggestion
+ "This is syntactic sugar that auto-generates the JAAS config string. "
+ "Warning: this option stores plaintext credentials and should be handled as sensitive configuration. "
+ "Avoid placing it in shared config files or exposing it through logs, config listing APIs, or UIs.");
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java:
##########
@@ -783,6 +785,119 @@ void testListRebalanceProgress() throws Exception {
});
}
+ @Test
+ void testAddAndDeleteUser() throws Exception {
+ String bobBootstrapServers =
+ String.join(
+ ",",
+ FLUSS_CLUSTER_EXTENSION
+ .getClientConfig("CLIENT")
+ .get(ConfigOptions.BOOTSTRAP_SERVERS));
+ String bobCatalog = "bob_catalog";
+ String createCatalogDDL =
+ String.format(
+ "create catalog %s with ("
+ + "'type' = 'fluss', "
+ + "'bootstrap.servers' = '%s', "
+ + "'client.security.protocol' = 'sasl', "
+ + "'client.security.sasl.mechanism' = 'PLAIN', "
+ + "'client.security.sasl.username' = 'bob', "
+ + "'client.security.sasl.password' = 'bob_pass'"
+ + ")",
+ bobCatalog, bobBootstrapServers);
+
+ assertThatThrownBy(() -> tEnv.executeSql(createCatalogDDL).await())
+ .hasMessageContaining("Invalid username or password");
+
+ // Step 1: Add user "bob" via append_cluster_configs
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.append_cluster_configs('%s', 'bob:bob_pass')",
+ CATALOG_NAME, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ assertThat(results.get(0).getField(0))
+ .asString()
+ .contains("Successfully appended")
+ .contains(ConfigOptions.SERVER_SASL_USERS.key());
+ }
+
+ // Verify user "bob" was added
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.get_cluster_configs('%s')",
+ CATALOG_NAME, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ assertThat((String) results.get(0).getField(1)).contains("bob:bob_pass");
+ }
+
+ // Verify "bob" can authenticate by creating a catalog with bob's credentials
+ tEnv.executeSql(createCatalogDDL).await();
+
+ // Grant bob DESCRIBE permission on cluster so bob can query configs
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.add_acl('CLUSTER', 'ALLOW', 'User:bob', 'DESCRIBE', '*')",
+ CATALOG_NAME))
+ .await();
+
+ // Bob should be able to get cluster configs
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.get_cluster_configs('%s')",
+ bobCatalog, ConfigOptions.SERVER_SASL_USERS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ }
+ tEnv.executeSql("drop catalog " + bobCatalog);
Review Comment:
This IT adds cluster state (catalog + ACL) but cleanup is not protected by a `finally` block. If an assertion or intermediate call fails after creating the catalog or adding the ACL, the test can leak state and make subsequent tests flaky. Wrap catalog creation/ACL grant in `try`/`finally` to reliably drop the catalog and remove the ACL.
```suggestion
boolean bobCatalogCreated = false;
boolean bobDescribeAclAdded = false;
try {
// Verify "bob" can authenticate by creating a catalog with bob's credentials
tEnv.executeSql(createCatalogDDL).await();
bobCatalogCreated = true;
// Grant bob DESCRIBE permission on cluster so bob can query configs
tEnv.executeSql(
String.format(
"Call %s.sys.add_acl('CLUSTER', 'ALLOW', 'User:bob', 'DESCRIBE', '*')",
CATALOG_NAME))
.await();
bobDescribeAclAdded = true;
// Bob should be able to get cluster configs
try (CloseableIterator<Row> resultIterator =
tEnv.executeSql(
String.format(
"Call %s.sys.get_cluster_configs('%s')",
bobCatalog, ConfigOptions.SERVER_SASL_USERS.key()))
.collect()) {
List<Row> results = CollectionUtil.iteratorToList(resultIterator);
assertThat(results).hasSize(1);
}
} finally {
if (bobDescribeAclAdded) {
tEnv.executeSql(
String.format(
"Call %s.sys.remove_acl('CLUSTER', 'ALLOW', 'User:bob', 'DESCRIBE', '*')",
CATALOG_NAME))
.await();
}
if (bobCatalogCreated) {
tEnv.executeSql("drop catalog " + bobCatalog).await();
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Manage users for sasl/plain authentication via cluster properties. [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #3047:
URL: https://github.com/apache/fluss/pull/3047#issuecomment-4924002411
@loserwang1024 I pushed several commits to enhance the code, please take a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org