Skip to content

[server] Only super.users can modify configuration of super.user info. #4229

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wuchong merged 2 commits into from
Sep 8, 2026
Merged

[server] Only super.users can modify configuration of super.user info. #4229

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ef3c747
[server] Only super.users can modify configuration of super.user info.
Sep 3, 2026
6fd0974
modified based on jark's advice.
Sep 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions fluss-common/src/main/java/org/apache/fluss/config/cluster/ServerReconfigurable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,9 @@

import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.ConfigException;
import org.apache.fluss.security.acl.FlussPrincipal;

import javax.annotation.Nullable;

/** Server Reconfigurable Interface which can dynamically respond to configuration changes. */
public interface ServerReconfigurable {
Expand All @@ -43,6 +46,22 @@ public interface ServerReconfigurable {
*/
void validate(Configuration newConfig) throws ConfigException;

/**
* Validates the provided configuration on behalf of the requester, which allows implementations
* to additionally reject changes the requester is not allowed to make. The default
* implementation ignores the requester and delegates to {@link #validate(Configuration)}.
*
* @param newConfig the new configuration, see {@link #validate(Configuration)}
* @param requester the principal that requested the change, or null if the reconfiguration is
* triggered by the server itself
* @throws ConfigException if the configuration is invalid or cannot be applied to this
* component
*/
default void validate(Configuration newConfig, @Nullable FlussPrincipal requester)
throws ConfigException {
validate(newConfig);
}

/**
* Reconfigures the component with the provided configuration.
*
Expand Down
14 changes: 14 additions & 0 deletions fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,10 @@
import org.apache.fluss.annotation.PublicEvolving;

import java.security.Principal;
import java.util.Arrays;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/**
* Represents a security principal in Fluss, defined by a {@code name} and {@code type}.
Expand Down Expand Up @@ -56,6 +59,17 @@ public FlussPrincipal(String name, String type) {
this.type = type;
}

/**
* Parses principals from a semicolon separated list of {@code <type>:<name>} pairs, e.g. {@code
* User:root;Group:admins}.
*/
public static Set<FlussPrincipal> parsePrincipals(String principals) {
return Arrays.stream(principals.split(";"))
.map(principal -> principal.trim().split(":"))
.map(principal -> new FlussPrincipal(principal[1], principal[0]))
.collect(Collectors.toSet());
}

@Override
public String getName() {
return name;
Expand Down
110 changes: 105 additions & 5 deletions ...uss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.MemorySize;
import org.apache.fluss.exception.AuthorizationException;
import org.apache.fluss.exception.NoRebalanceInProgressException;
import org.apache.fluss.exception.SecurityDisabledException;
import org.apache.fluss.metadata.DataLakeFormat;
Expand Down Expand Up @@ -866,7 +867,7 @@ void testAddAndDeleteUser() throws Exception {
assertThat(results).hasSize(1);
assertThat(results.stream().map(Row::toString).collect(Collectors.toList()))
.containsExactly(
"+I[security.sasl.plain.credentials, root:******,guest:******,bob:******, DYNAMIC_SERVER_CONFIG]");
"+I[security.sasl.plain.credentials, root:******,super:******,bob:******, DYNAMIC_SERVER_CONFIG]");
}

// Verify "bob" can authenticate by creating a catalog with bob's credentials.
Expand Down Expand Up @@ -894,8 +895,93 @@ void testAddAndDeleteUser() throws Exception {
List<Row> results = CollectionUtil.iteratorToList(resultIterator);
assertThat(results.stream().map(Row::toString).collect(Collectors.toList()))
.containsExactly(
"+I[security.sasl.plain.credentials, root:******,guest:******,bob:******, DYNAMIC_SERVER_CONFIG]");
"+I[security.sasl.plain.credentials, root:******,super:******,bob:******, DYNAMIC_SERVER_CONFIG]");
}

String credentialsKey = ConfigOptions.SERVER_SASL_CREDENTIALS.key();
String credentialsWithAlice = "root:password,super:passwords,bob:bob_pass,alice:alice_pass";
String credentialsWithChangedSuperUser =
"root:password,super:new-password,bob:bob_pass,alice:alice_pass";

// Security-related cluster configs require ALL rather than ALTER.
tEnv.executeSql(
String.format(
"Call %s.sys.add_acl('CLUSTER', 'ALLOW', 'User:bob', 'ALTER', '*')",
CATALOG_NAME))
.await();
assertThatThrownBy(
() ->
tEnv.executeSql(
String.format(
"Call %s.sys.set_cluster_configs('%s', '5min', '%s', '%s')",
bobCatalog,
ConfigOptions.KV_SNAPSHOT_INTERVAL.key(),
credentialsKey,
credentialsWithAlice))
.await())
.rootCause()
.isInstanceOf(AuthorizationException.class)
.hasMessageContaining("operate ALL");

// ALL allows Bob to alter ordinary credentials, but not super-user credentials.
// The super users of this cluster are "root" and "super" (see initConfig()), so even
// though Bob is granted CLUSTER ALL, he may only add/remove/change ordinary users like
// "alice"; changing or removing the "super" account is rejected.
tEnv.executeSql(
String.format(
"Call %s.sys.add_acl('CLUSTER', 'ALLOW', 'User:bob', 'ALL', '*')",
CATALOG_NAME))
.await();
tEnv.executeSql(
String.format(
"Call %s.sys.append_cluster_configs('%s', 'alice:alice_pass')",
bobCatalog, credentialsKey))
.await();
// Bob (CLUSTER ALL, but not a super user) cannot change the super user's password.
assertThatThrownBy(
() ->
tEnv.executeSql(
String.format(
"Call %s.sys.set_cluster_configs('%s', '%s')",
bobCatalog,
credentialsKey,
credentialsWithChangedSuperUser))
.await())
.rootCause()
.isInstanceOf(AuthorizationException.class)
.hasMessageContaining(
"cannot modify credentials belonging to users in 'super.users'");
// Nor can he remove the super user account.
assertThatThrownBy(
() ->
tEnv.executeSql(
String.format(
"Call %s.sys.subtract_cluster_configs('%s', 'super:passwords')",
bobCatalog, credentialsKey))
.await())
.rootCause()
.isInstanceOf(AuthorizationException.class)
.hasMessageContaining(
"cannot modify credentials belonging to users in 'super.users'");

// A super user may alter another configured super user's credentials: "root" is a super
// user, so it can change the password of the "super" account and restore it afterwards.
tEnv.executeSql(
String.format(
"Call %s.sys.set_cluster_configs('%s', '%s')",
CATALOG_NAME, credentialsKey, credentialsWithChangedSuperUser))
.await();
tEnv.executeSql(
String.format(
"Call %s.sys.set_cluster_configs('%s', '%s')",
CATALOG_NAME, credentialsKey, credentialsWithAlice))
.await();
tEnv.executeSql(
String.format(
"Call %s.sys.subtract_cluster_configs('%s', 'alice:alice_pass')",
bobCatalog, credentialsKey))
.await();

tEnv.executeSql("drop catalog " + bobCatalog);

// Step 2: Delete user "bob" via subtract_cluster_configs
Expand All @@ -916,7 +1002,7 @@ void testAddAndDeleteUser() throws Exception {
// After subtracting the only dynamically-added entry, the config may be empty
assertThat(results.stream().map(Row::toString).collect(Collectors.toList()))
.containsExactly(
"+I[security.sasl.plain.credentials, root:******,guest:******, DYNAMIC_SERVER_CONFIG]");
"+I[security.sasl.plain.credentials, root:******,super:******, DYNAMIC_SERVER_CONFIG]");
}

// Verify "bob" can no longer authenticate.
Expand All @@ -932,6 +1018,16 @@ void testAddAndDeleteUser() throws Exception {
"Call %s.sys.drop_acl('CLUSTER', 'ALLOW', 'User:bob', 'DESCRIBE', '*')",
CATALOG_NAME))
.await();
tEnv.executeSql(
String.format(
"Call %s.sys.drop_acl('CLUSTER', 'ALLOW', 'User:bob', 'ALTER', '*')",
CATALOG_NAME))
.await();
tEnv.executeSql(
String.format(
"Call %s.sys.drop_acl('CLUSTER', 'ALLOW', 'User:bob', 'ALL', '*')",
CATALOG_NAME))
.await();
// Try to append a map entry with the same key as the existing "root" entry
assertThatThrownBy(
() ->
Expand Down Expand Up @@ -1073,9 +1169,13 @@ private static Configuration initConfig() {
// set security information.
conf.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(), "CLIENT:sasl");
conf.setString("security.sasl.enabled.mechanisms", "plain");
// Two users are configured statically, and both of them are super users:
// "root" is the user the tests' default catalog connects with, and "super" is a second
// super user which is only used to verify that even a CLUSTER ALL grant does not allow
// altering the credentials of a super user account.
conf.setString(
ConfigOptions.SERVER_SASL_CREDENTIALS.key(), "root:password,guest:passwords");
conf.set(ConfigOptions.SUPER_USERS, "User:root");
ConfigOptions.SERVER_SASL_CREDENTIALS.key(), "root:password,super:passwords");
conf.set(ConfigOptions.SUPER_USERS, "User:root;User:super");
conf.set(ConfigOptions.AUTHORIZER_ENABLED, true);
return conf;
}
Expand Down
89 changes: 84 additions & 5 deletions fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,19 +21,26 @@
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.cluster.ServerReconfigurable;
import org.apache.fluss.exception.AuthorizationException;
import org.apache.fluss.exception.ConfigException;
import org.apache.fluss.rpc.RpcGatewayService;
import org.apache.fluss.rpc.protocol.ApiManager;
import org.apache.fluss.rpc.protocol.NetworkProtocolPlugin;
import org.apache.fluss.security.acl.FlussPrincipal;
import org.apache.fluss.security.auth.AuthenticationFactory;
import org.apache.fluss.security.auth.PlainTextAuthenticationPlugin;
import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelHandler;

import javax.annotation.Nullable;

import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

Expand Down Expand Up @@ -65,6 +72,9 @@ public class FlussProtocolPlugin implements NetworkProtocolPlugin, ServerReconfi
private final List<String> listeners;
private final RequestsMetrics requestsMetrics;
private Configuration conf;
private Set<FlussPrincipal> superUsers;
private boolean principalIgnoreCase;

/** Initial credentials from `security.sasl.plain.jaas.config`. */
private Map<String, String> initialPlainCredentialsFromJaasConfig;

Expand All @@ -86,6 +96,8 @@ public String name() {
@Override
public void setup(Configuration conf) {
this.conf = new Configuration(conf);
this.principalIgnoreCase = this.conf.get(ConfigOptions.SECURITY_ACL_PRINCIPAL_IGNORE_CASE);
this.superUsers = parseSuperUsers(this.conf);
this.initialPlainCredentialsFromJaasConfig = parseCredentialsFromJaasConfig(conf);
enrichWithJaasConfig(conf);
}
Expand Down Expand Up @@ -138,6 +150,13 @@ public void validate(Configuration newConfig) throws ConfigException {
generateMergedJaasConfig(newCredentials);
}

@Override
public void validate(Configuration newConfig, @Nullable FlussPrincipal requester)
throws ConfigException {
authorizeSuperUserCredentialChanges(readPlainCredentials(newConfig), requester);
validate(newConfig);
}

@Override
public void reconfigure(Configuration newConfig) throws ConfigException {
enrichWithJaasConfig(newConfig);
Expand Down Expand Up @@ -209,11 +228,7 @@ private static void validatePassword(int index, String username, String password
* @return the generated JAAS config string
*/
private String generateMergedJaasConfig(Map<String, String> newCredentials) {
Map<String, String> mergedCredentials =
new LinkedHashMap<>(initialPlainCredentialsFromJaasConfig);
if (newCredentials != null) {
mergedCredentials.putAll(newCredentials);
}
Map<String, String> mergedCredentials = mergePlainCredentials(newCredentials);

StringBuilder sb =
new StringBuilder(
Expand All @@ -225,6 +240,70 @@ private String generateMergedJaasConfig(Map<String, String> newCredentials) {
return sb.toString();
}

private Map<String, String> mergePlainCredentials(Map<String, String> plainCredentials) {
Map<String, String> mergedCredentials =
new LinkedHashMap<>(initialPlainCredentialsFromJaasConfig);
if (plainCredentials != null) {
mergedCredentials.putAll(plainCredentials);
}
return mergedCredentials;
}

/**
* Rejects the change if the requester is not a configured super user but the credentials of a
* configured super user would be added, removed or modified.
*/
private void authorizeSuperUserCredentialChanges(
@Nullable Map<String, String> newCredentials, @Nullable FlussPrincipal requester) {
if (requester == null || isSuperUser(requester)) {
return;
}

if (!Objects.equals(
superUserCredentials(currentPlainCredentials),
superUserCredentials(newCredentials))) {
throw new AuthorizationException(
String.format(
"Principal %s cannot modify credentials belonging to users in 'super.users', "
+ "the requester must itself be a super user.",
requester));
}
}

/** Returns the merged credentials whose username matches the name of a configured superuser. */
private Map<String, String> superUserCredentials(@Nullable Map<String, String> credentials) {
Map<String, String> superUserCredentials = new HashMap<>();
mergePlainCredentials(credentials)
.forEach(
(user, password) -> {
if (isSuperUserName(user)) {
superUserCredentials.put(user, password);
}
});
return superUserCredentials;
}

private boolean isSuperUser(FlussPrincipal principal) {
return superUsers.stream()
.anyMatch(superUser -> superUser.matches(principal, principalIgnoreCase));
}

private boolean isSuperUserName(String username) {
return superUsers.stream()
.anyMatch(
superUser ->
principalIgnoreCase
? username.equalsIgnoreCase(superUser.getName())
: username.equals(superUser.getName()));
}

private static Set<FlussPrincipal> parseSuperUsers(Configuration configuration) {
return configuration
.getOptional(ConfigOptions.SUPER_USERS)
.map(FlussPrincipal::parsePrincipals)
.orElse(Collections.emptySet());
}

private static Map<String, String> parseCredentialsFromJaasConfig(Configuration configuration) {
Map<String, String> credentials = new LinkedHashMap<>();
String existingJaas = configuration.getString(ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG);
Expand Down
Loading
Loading