diff --git a/fluss-common/src/main/java/org/apache/fluss/config/cluster/ServerReconfigurable.java b/fluss-common/src/main/java/org/apache/fluss/config/cluster/ServerReconfigurable.java index 882263b68bd..a93a70645cd 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/cluster/ServerReconfigurable.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/cluster/ServerReconfigurable.java @@ -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 { @@ -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. * diff --git a/fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java b/fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java index 7a60e9c7b6f..5e2ce93917e 100644 --- a/fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java +++ b/fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java @@ -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}. @@ -56,6 +59,17 @@ public FlussPrincipal(String name, String type) { this.type = type; } + /** + * Parses principals from a semicolon separated list of {@code :} pairs, e.g. {@code + * User:root;Group:admins}. + */ + public static Set 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; diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java index f1c71f0c26c..4f03464fc2f 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java @@ -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; @@ -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 @@ -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( () -> @@ -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"); conf.set(ConfigOptions.AUTHORIZER_ENABLED, true); return conf; } 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..243198ef4d0 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 @@ -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; @@ -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_=""} entries in JAAS config strings. */ private static final Pattern JAAS_USER_PATTERN = Pattern.compile("user_(\\w+)=\"([^\"]*)\""); @@ -65,6 +72,9 @@ public class FlussProtocolPlugin implements NetworkProtocolPlugin, ServerReconfi private final List listeners; private final RequestsMetrics requestsMetrics; private Configuration conf; + private Set superUsers; + private boolean principalIgnoreCase; + /** Initial credentials from `security.sasl.plain.jaas.config`. */ private Map initialPlainCredentialsFromJaasConfig; @@ -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); } @@ -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); @@ -209,11 +228,7 @@ private static void validatePassword(int index, String username, String password * @return the generated JAAS config string */ private String generateMergedJaasConfig(Map newCredentials) { - Map mergedCredentials = - new LinkedHashMap<>(initialPlainCredentialsFromJaasConfig); - if (newCredentials != null) { - mergedCredentials.putAll(newCredentials); - } + Map mergedCredentials = mergePlainCredentials(newCredentials); StringBuilder sb = new StringBuilder( @@ -225,6 +240,58 @@ private String generateMergedJaasConfig(Map newCredentials) { return sb.toString(); } + private Map mergePlainCredentials(Map plainCredentials) { + Map 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 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."); + } + } + + /** Returns the merged credentials that belong to a configured super user. */ + private Map superUserCredentials(@Nullable Map credentials) { + Map superUserCredentials = new LinkedHashMap<>(); + mergePlainCredentials(credentials) + .forEach( + (user, password) -> { + if (isSuperUser(new FlussPrincipal(user, USER_PRINCIPAL_TYPE))) { + superUserCredentials.put(user, password); + } + }); + return superUserCredentials; + } + + private boolean isSuperUser(FlussPrincipal principal) { + return superUsers.stream() + .anyMatch(superUser -> superUser.matches(principal, principalIgnoreCase)); + } + + private static Set parseSuperUsers(Configuration configuration) { + return configuration + .getOptional(ConfigOptions.SUPER_USERS) + .map(FlussPrincipal::parsePrincipals) + .orElse(Collections.emptySet()); + } + private static Map parseCredentialsFromJaasConfig(Configuration configuration) { Map credentials = new LinkedHashMap<>(); String existingJaas = configuration.getString(ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG); diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/FlussProtocolPluginTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/FlussProtocolPluginTest.java new file mode 100644 index 00000000000..dd3b656f853 --- /dev/null +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/FlussProtocolPluginTest.java @@ -0,0 +1,95 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.rpc.netty.server; + +import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.AuthorizationException; +import org.apache.fluss.metrics.groups.MetricGroup; +import org.apache.fluss.metrics.util.NOPMetricsGroup; +import org.apache.fluss.security.acl.FlussPrincipal; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link FlussProtocolPlugin}. */ +class FlussProtocolPluginTest { + + private static final FlussPrincipal ROOT = new FlussPrincipal("root", "User"); + private static final FlussPrincipal OPERATOR = new FlussPrincipal("operator", "User"); + + @Test + void testAuthorizeSuperUserCredentialChanges() { + FlussProtocolPlugin plugin = createPlugin(false); + + // a non-super user may change ordinary credentials + Configuration addedAlice = credentials("root:root-pass,operator:new-pass,alice:alice-pass"); + assertThatCode(() -> plugin.validate(addedAlice, OPERATOR)).doesNotThrowAnyException(); + + // but not the credentials of a configured super user + Configuration changedRoot = credentials("root:new-root-pass,operator:operator-pass"); + assertThatThrownBy(() -> plugin.validate(changedRoot, OPERATOR)) + .isInstanceOf(AuthorizationException.class) + .hasMessageContaining("Only configured super users may alter credentials"); + + // a super user may, and so may the server itself + assertThatCode(() -> plugin.validate(changedRoot, ROOT)).doesNotThrowAnyException(); + assertThatCode(() -> plugin.validate(changedRoot, null)).doesNotThrowAnyException(); + } + + @Test + void testAuthorizeSuperUserCredentialChangesIgnoringCase() { + FlussProtocolPlugin plugin = createPlugin(true); + + // the super user lookup ignores the case of the principal name and type + Configuration changedRoot = credentials("root:new-root-pass,operator:operator-pass"); + assertThatCode(() -> plugin.validate(changedRoot, new FlussPrincipal("ROOT", "USER"))) + .doesNotThrowAnyException(); + + // renaming a super user by case only still changes super user credentials + Configuration renamedRoot = credentials("ROOT:root-pass,operator:operator-pass"); + assertThatThrownBy(() -> plugin.validate(renamedRoot, OPERATOR)) + .isInstanceOf(AuthorizationException.class); + } + + private static FlussProtocolPlugin createPlugin(boolean principalIgnoreCase) { + Configuration configuration = credentials("root:root-pass,operator:operator-pass"); + configuration.set(ConfigOptions.SUPER_USERS, "User:root"); + configuration.set(ConfigOptions.SECURITY_ACL_PRINCIPAL_IGNORE_CASE, principalIgnoreCase); + + MetricGroup metricGroup = NOPMetricsGroup.newInstance(); + FlussProtocolPlugin plugin = + new FlussProtocolPlugin( + ServerType.COORDINATOR, + Collections.emptyList(), + RequestsMetrics.createCoordinatorServerRequestMetrics(metricGroup)); + plugin.setup(configuration); + return plugin; + } + + private static Configuration credentials(String credentials) { + Configuration configuration = new Configuration(); + configuration.setString(ConfigOptions.SERVER_SASL_CREDENTIALS.key(), credentials); + return configuration; + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java b/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java index dec53519b2a..824de74a8a9 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java @@ -27,6 +27,7 @@ import org.apache.fluss.config.cluster.ServerReconfigurable; import org.apache.fluss.config.provider.ConfigProviders; import org.apache.fluss.exception.ConfigException; +import org.apache.fluss.security.acl.FlussPrincipal; import org.apache.fluss.server.authorizer.ZkNodeChangeNotificationWatcher; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.ZkData.ConfigEntityChangeNotificationSequenceZNode; @@ -36,6 +37,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -88,7 +91,7 @@ public void startup() throws Exception { try { configChangeListener.start(); Map entityConfigs = zooKeeperClient.fetchEntityConfig(); - dynamicServerConfig.updateDynamicConfig(entityConfigs, true); + dynamicServerConfig.updateDynamicConfig(entityConfigs, true, null); } catch (Exception e) { LOG.error("Failed to update dynamic configs from zookeeper", e); } @@ -135,7 +138,7 @@ public void pauseListening() { public void resumeListening() throws Exception { listeningEnabled = true; Map entityConfigs = zooKeeperClient.fetchEntityConfig(); - dynamicServerConfig.updateDynamicConfig(entityConfigs, true); + dynamicServerConfig.updateDynamicConfig(entityConfigs, true, null); } public List describeConfigs() { @@ -168,9 +171,19 @@ public List describeConfigs() { } public void alterConfigs(List clusterConfigChanges) throws Exception { + alterConfigs(clusterConfigChanges, null); + } + + /** + * Alters the cluster configs on behalf of the requester, which is null if the change is + * triggered by the server itself. + */ + public void alterConfigs( + List clusterConfigChanges, @Nullable FlussPrincipal requester) + throws Exception { Map persistentProps = zooKeeperClient.fetchEntityConfig(); prepareIncrementalConfigs(clusterConfigChanges, persistentProps); - alterServerConfigs(persistentProps); + alterServerConfigs(persistentProps, requester); } private void prepareIncrementalConfigs( @@ -359,8 +372,9 @@ private String getExistingConfigValue(Map dynamicConfigs, String } @VisibleForTesting - protected void alterServerConfigs(Map configsProps) throws Exception { - dynamicServerConfig.updateDynamicConfig(configsProps, false); + protected void alterServerConfigs( + Map configsProps, @Nullable FlussPrincipal requester) throws Exception { + dynamicServerConfig.updateDynamicConfig(configsProps, false, requester); // Apply to zookeeper only after verification. zooKeeperClient.upsertServerEntityConfig(configsProps); @@ -400,7 +414,7 @@ public void processNotification(byte[] notification) throws Exception { } Map entityConfig = zooKeeperClient.fetchEntityConfig(); - dynamicServerConfig.updateDynamicConfig(entityConfig, true); + dynamicServerConfig.updateDynamicConfig(entityConfig, true, null); } } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java index 98d721e152e..217253c2a23 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java @@ -25,12 +25,15 @@ import org.apache.fluss.config.cluster.ConfigValidator; import org.apache.fluss.config.cluster.ServerReconfigurable; import org.apache.fluss.exception.ConfigException; +import org.apache.fluss.security.acl.FlussPrincipal; import org.apache.fluss.server.config.ConfigRedactor; import org.apache.fluss.server.config.ConfigRedactors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -180,9 +183,12 @@ private static boolean isPlainJaasConfig(String configKey) { * Update the dynamic configuration and apply to registered ServerReconfigurable. If skipping * error config, only the error one will be ignored. */ - void updateDynamicConfig(Map newDynamicConfigs, boolean skipErrorConfig) + void updateDynamicConfig( + Map newDynamicConfigs, + boolean skipErrorConfig, + @Nullable FlussPrincipal requester) throws Exception { - inWriteLock(lock, () -> updateCurrentConfig(newDynamicConfigs, skipErrorConfig)); + inWriteLock(lock, () -> updateCurrentConfig(newDynamicConfigs, skipErrorConfig, requester)); } Map getDynamicConfigs() { @@ -206,7 +212,10 @@ boolean isAllowedConfig(String key) { return false; } - private void updateCurrentConfig(Map newDynamicConfigs, boolean skipErrorConfig) + private void updateCurrentConfig( + Map newDynamicConfigs, + boolean skipErrorConfig, + @Nullable FlussPrincipal requester) throws Exception { // Compute effective config changes (merge with initial configs) Map effectiveChanges = @@ -225,7 +234,7 @@ private void updateCurrentConfig(Map newDynamicConfigs, boolean Configuration newConfig = Configuration.fromMap(newConfigMap); // Apply changes to all registered ServerReconfigurable instances - applyToServerReconfigurables(newConfig, skipErrorConfig); + applyToServerReconfigurables(newConfig, skipErrorConfig, requester); // Update internal state updateInternalState(newConfig, newConfigMap, newDynamicConfigs); @@ -444,9 +453,11 @@ private void validateSingleConfig(String configKey, String oldValueStr, String n * * @param newConfig new configuration to apply * @param skipErrorConfig whether to skip errors + * @param requester the principal that requested the change, or null if triggered by the server * @throws Exception if apply fails and skipErrorConfig is false */ - private void applyToServerReconfigurables(Configuration newConfig, boolean skipErrorConfig) + private void applyToServerReconfigurables( + Configuration newConfig, boolean skipErrorConfig, @Nullable FlussPrincipal requester) throws Exception { Configuration oldConfig = currentConfig; Set appliedSet = new HashSet<>(); @@ -454,7 +465,7 @@ private void applyToServerReconfigurables(Configuration newConfig, boolean skipE // Validate all first for (ServerReconfigurable reconfigurable : serverReconfigures.values()) { try { - reconfigurable.validate(newConfig); + reconfigurable.validate(newConfig, requester); } catch (ConfigException e) { LOG.error( "Validation failed for {}: {}", diff --git a/fluss-server/src/main/java/org/apache/fluss/server/authorizer/DefaultAuthorizer.java b/fluss-server/src/main/java/org/apache/fluss/server/authorizer/DefaultAuthorizer.java index 0c38f5d02b2..f6e0739ac85 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/authorizer/DefaultAuthorizer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/authorizer/DefaultAuthorizer.java @@ -594,17 +594,7 @@ private VersionedAcls getAclsFromZk(Resource resource) throws Exception { private static Set parseSuperUsers(Configuration configuration) { return configuration .getOptional(ConfigOptions.SUPER_USERS) - .map( - config -> - Arrays.stream(config.split(";")) - .map(String::trim) - .map( - user -> { - String[] userInfo = user.split(":"); - return new FlussPrincipal( - userInfo[1], userInfo[0]); - }) - .collect(Collectors.toSet())) + .map(FlussPrincipal::parsePrincipals) .orElse(Collections.emptySet()); } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java index 94e837ce653..31238cee297 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java @@ -242,6 +242,7 @@ public final class CoordinatorService extends RpcServiceBase implements CoordinatorGateway { private static final Logger LOG = LoggerFactory.getLogger(CoordinatorService.class); + private static final String SECURITY_CONFIG_KEY_PREFIX = "security."; private final int defaultBucketNumber; private final int defaultReplicationFactor; @@ -1457,10 +1458,6 @@ public CompletableFuture alterClusterConfigs( return CompletableFuture.completedFuture(new AlterClusterConfigsResponse()); } - if (authorizer != null) { - authorizer.authorize(currentSession(), OperationType.ALTER, Resource.cluster()); - } - List serverConfigChanges = infos.stream() .map( @@ -1472,11 +1469,27 @@ public CompletableFuture alterClusterConfigs( : null, AlterConfigOpType.from((byte) info.getOpType()))) .collect(Collectors.toList()); + + Session session = currentSession(); + if (authorizer != null) { + // altering security related configs (e.g. super user credentials) requires the full + // cluster permission instead of ALTER only + boolean alterSecurityConfigs = + serverConfigChanges.stream() + .anyMatch( + config -> config.key().startsWith(SECURITY_CONFIG_KEY_PREFIX)); + authorizer.authorize( + session, + alterSecurityConfigs ? OperationType.ALL : OperationType.ALTER, + Resource.cluster()); + } + FlussPrincipal requester = session.isInternal() ? null : session.getPrincipal(); + AccessContextEvent accessContextEvent = new AccessContextEvent<>( (context) -> { try { - dynamicConfigManager.alterConfigs(serverConfigChanges); + dynamicConfigManager.alterConfigs(serverConfigChanges, requester); future.complete(new AlterClusterConfigsResponse()); } catch (ApiException e) { future.completeExceptionally(e);