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

Open
loserwang1024 wants to merge 1 commit into
base: main
Choose a base branch
from
+345 −34
Open

[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
1 commit
Select commit
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
90 changes: 89 additions & 1 deletion ...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 @@ -896,6 +897,83 @@ void testAddAndDeleteUser() throws Exception {
.containsExactly(
"+I[security.sasl.plain.credentials, root:******,guest:******,bob:******, DYNAMIC_SERVER_CONFIG]");
}

String credentialsKey = ConfigOptions.SERVER_SASL_CREDENTIALS.key();
String credentialsWithAlice = "root:password,guest:passwords,bob:bob_pass,alice:alice_pass";
String credentialsWithChangedGuest =
"root:password,guest: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.
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();
assertThatThrownBy(
() ->
tEnv.executeSql(
String.format(
"Call %s.sys.set_cluster_configs('%s', '%s')",
bobCatalog,
credentialsKey,
credentialsWithChangedGuest))
.await())
.rootCause()
.isInstanceOf(AuthorizationException.class)
.hasMessageContaining("Only configured super users may alter credentials");
assertThatThrownBy(
() ->
tEnv.executeSql(
String.format(
"Call %s.sys.subtract_cluster_configs('%s', 'guest:passwords')",
bobCatalog, credentialsKey))
.await())
.rootCause()
.isInstanceOf(AuthorizationException.class)
.hasMessageContaining("Only configured super users may alter credentials");

// A super user may alter another configured super user's credentials.
tEnv.executeSql(
String.format(
"Call %s.sys.set_cluster_configs('%s', '%s')",
CATALOG_NAME, credentialsKey, credentialsWithChangedGuest))
.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 Down Expand Up @@ -932,6 +1010,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 @@ -1075,7 +1163,7 @@ private static Configuration initConfig() {
conf.setString("security.sasl.enabled.mechanisms", "plain");
conf.setString(
ConfigOptions.SERVER_SASL_CREDENTIALS.key(), "root:password,guest:passwords");
conf.set(ConfigOptions.SUPER_USERS, "User:root");
conf.set(ConfigOptions.SUPER_USERS, "User:root;User:guest");

@wuchong wuchong Sep 3, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we use super instead of guest to represent the super user? The current test code is confusing, particularly when encountering the restriction on modifying the guest password. Additionally, please add comments near the test code to clarify which users are designated as super users and to explain that the CLUSTER ALL permission does not allow modifications to super user accounts.

conf.set(ConfigOptions.AUTHORIZER_ENABLED, true);
return conf;
}
Expand Down
77 changes: 72 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,25 @@
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.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 All @@ -42,6 +48,7 @@ public class FlussProtocolPlugin implements NetworkProtocolPlugin, ServerReconfi

private static final String PLAIN_CREDENTIALS_CONFIG =
ConfigOptions.SERVER_SASL_CREDENTIALS.key();
private static final String USER_PRINCIPAL_TYPE = "User";

/** Pattern to match {@code user_<username>="<password>"} entries in JAAS config strings. */
private static final Pattern JAAS_USER_PATTERN = Pattern.compile("user_(\\w+)=\"([^\"]*)\"");
Expand All @@ -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,58 @@ 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(
"Only configured super users may alter credentials of configured super users.");
Comment on lines +265 to +266

@wuchong wuchong Sep 3, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
throw new AuthorizationException(
"Only configured super users may alter credentials of configured super users.");
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));

Improve the exception to make it more clear.

}
}

/** Returns the merged credentials that belong to a configured super user. */
private Map<String, String> superUserCredentials(@Nullable Map<String, String> credentials) {
Map<String, String> superUserCredentials = new LinkedHashMap<>();

@wuchong wuchong Sep 3, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we need LinkedHashMap here? I think we don't need the original order to compare the super user credentials.

mergePlainCredentials(credentials)
.forEach(
(user, password) -> {
if (isSuperUser(new FlussPrincipal(user, USER_PRINCIPAL_TYPE))) {

@wuchong wuchong Sep 3, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is the User type hard-coded here? According to the description of super.users, the principal_type can be configured to any type, not just User. Shouldn't we simply compare usernames instead? After all, we assume that usernames within the security.sasl.plain.credentials are unique.

superUserCredentials.put(user, password);
}
});
return superUserCredentials;
}

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

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