[security] Support SASL/PLAIN proxy authorization

fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java

@@ -81,6 +81,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.time.Duration;
@@ -349,6 +350,51 @@ void testAlterDatabase() throws Exception {
         assertThat(rootAdmin.databaseExists("test-database2").get()).isTrue();
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = {"guest", "tenant_a_guest"})
+    void testImpersonation(String targetUser) throws Exception {
+        Configuration proxyConf = proxyConfig("proxy", "password3", targetUser);
+        FlussPrincipal targetPrincipal = new FlussPrincipal(targetUser, "User");
+        try (Connection proxyConn = ConnectionFactory.createConnection(proxyConf);
+                Admin proxyAdmin = proxyConn.getAdmin()) {
+            assertThatThrownBy(() -> proxyAdmin.getTableInfo(DATA1_TABLE_PATH_PK).get())
+                    .cause()
+                    .isInstanceOf(AuthorizationException.class)
+                    .hasMessageContaining(
+                            "Principal " + targetPrincipal + " have no authorization");
+
+            // Grant DESCRIBE to the target, then the impersonated connection can access it.
+            List<AclBinding> aclBindings =
+                    Collections.singletonList(
+                            new AclBinding(
+                                    Resource.table(DATA1_TABLE_PATH_PK),
+                                    new AccessControlEntry(
+                                            targetPrincipal,
+                                            "*",
+                                            OperationType.DESCRIBE,
+                                            PermissionType.ALLOW)));
+            rootAdmin.createAcls(aclBindings).all().get();
+            FLUSS_CLUSTER_EXTENSION.waitUntilAuthenticationSync(aclBindings, true);
+            assertThat(proxyAdmin.getTableInfo(DATA1_TABLE_PATH_PK).get().getTablePath())
+                    .isEqualTo(DATA1_TABLE_PATH_PK);
+        }
+    }
+
+    @ParameterizedTest
+    @CsvSource({
+        "proxy, password3, tenant_b_guest",
+        "proxy, password3, xtenant_a_guest",
+        "proxy, password3, tenant_a_guest!",
+        "guest, password2, proxy",
+        "guest, password2, root"
+    })
+    void testImpersonationRejected(String username, String password, String targetUser) {
+        Configuration proxyConf = proxyConfig(username, password, targetUser);
+        assertThatThrownBy(() -> ConnectionFactory.createConnection(proxyConf))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("is not allowed to act as '" + targetUser + "'");
+    }
+
     @Test
     void testListDatabases() throws ExecutionException, InterruptedException {
         assertThat(guestAdmin.listDatabases().get())
@@ -1502,6 +1548,15 @@ void testTableExistsAuthorization() throws Exception {
         rootAdmin.dropTable(testTablePath, true).get();
     }
 
+    private static Configuration proxyConfig(String username, String password, String targetUser) {
+        Configuration conf = FLUSS_CLUSTER_EXTENSION.getClientConfig("CLIENT");
+        conf.set(ConfigOptions.CLIENT_SECURITY_PROTOCOL, "SASL");
+        conf.set(ConfigOptions.CLIENT_SASL_JAAS_USERNAME, username);
+        conf.set(ConfigOptions.CLIENT_SASL_JAAS_PASSWORD, password);
+        conf.set(ConfigOptions.CLIENT_SASL_AUTHORIZATION_ID, targetUser);
+        return conf;
+    }
+
     private static Configuration initConfig() {
         Configuration conf = new Configuration();
         conf.setInt(ConfigOptions.DEFAULT_REPLICATION_FACTOR, 3);
@@ -1520,7 +1575,11 @@ private static Configuration initConfig() {
                 "security.sasl.plain.jaas.config",
                 "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required "
                         + "    user_root=\"password\" "
-                        + "    user_guest=\"password2\";");
+                        + "    user_guest=\"password2\" "
+                        + "    user_proxy=\"password3\";");
+        conf.set(
+                ConfigOptions.SERVER_SASL_PROXY_USERS,
+                Collections.singletonMap("proxy", "guest;regex:^tenant_a_[a-z0-9_]+$"));
         conf.set(ConfigOptions.SUPER_USERS, "User:root");
         conf.set(ConfigOptions.AUTHORIZER_ENABLED, true);
         return conf;

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -700,6 +700,19 @@ public class ConfigOptions {
                                     + "from 'security.sasl.plain.credentials' when that credential map is set, "
                                     + "and can also be configured directly for compatibility.");
 
+    public static final ConfigOption<Map<String, String>> SERVER_SASL_PROXY_USERS =
+            key("security.sasl.proxy-users")
+                    .mapType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Users each authenticated SASL account may act as, "
+                                    + "e.g. `gateway:alice;bob,admin:*`. The wildcard `*` allows any target "
+                                    + "except a super user. Prefix a target rule with `regex:` for a "
+                                    + "full-string Java regular expression match, e.g. "
+                                    + "`gateway:'regex:^tenant_a_[a-z0-9_]+$'`. Other rules match literal "
+                                    + "usernames. Invalid patterns fail when loading the authentication "
+                                    + "configuration. Users in `super.users` cannot be impersonated.");
+
     // ------------------------------------------------------------------------
     //  Server-side TLS (transport encryption + mTLS) options
     // ------------------------------------------------------------------------
@@ -1993,7 +2006,7 @@ public class ConfigOptions {
                     .stringType()
                     .noDefaultValue()
                     .withDescription(
-                            "The password to use for client-side SASL JAAS authentication. "
+                            "The username to use for client-side SASL JAAS authentication. "
                                     + "This is used when the client connects to the Fluss cluster with SASL authentication enabled. "
                                     + "If not provided, the username will be read from the JAAS configuration string specified by `client.security.sasl.jaas.config`.");
 
@@ -2002,10 +2015,19 @@ public class ConfigOptions {
                     .stringType()
                     .noDefaultValue()
                     .withDescription(
-                            "The username to use for client-side SASL JAAS authentication. "
+                            "The password to use for client-side SASL JAAS authentication. "
                                     + "This is used when the client connects to the Fluss cluster with SASL authentication enabled. "
                                     + "If not provided, the password will be read from the JAAS configuration string specified by `client.security.sasl.jaas.config`.");
 
+    public static final ConfigOption<String> CLIENT_SASL_AUTHORIZATION_ID =
+            key("client.security.sasl.authorization-id")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The user to act as with SASL/PLAIN. If unset or empty, the authenticated "
+                                    + "account is used. A different user requires a server-side grant "
+                                    + "in 'security.sasl.proxy-users'.");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for Fluss Table
     // ------------------------------------------------------------------------

fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java

@@ -17,6 +17,7 @@
 
 package org.apache.fluss.security.acl;
 
+import org.apache.fluss.annotation.Internal;
 import org.apache.fluss.annotation.PublicEvolving;
 
 import java.security.Principal;
@@ -111,6 +112,12 @@ public boolean matches(FlussPrincipal other, boolean ignoreCase) {
         return equalsIgnoreCase(name, other.name) && equalsIgnoreCase(type, other.type);
     }
 
+    /** Checks a username against this principal's name regardless of principal type. */
+    @Internal
+    public boolean matchesName(String username, boolean ignoreCase) {
+        return ignoreCase ? equalsIgnoreCase(name, username) : Objects.equals(name, username);
+    }
+
     private static boolean equalsIgnoreCase(String a, String b) {
         if (a == b) {
             return true;

fluss-common/src/main/java/org/apache/fluss/security/auth/AuthenticationFactory.java

@@ -103,6 +103,8 @@ public static Map<String, Supplier<ServerAuthenticator>> loadServerAuthenticator
                             ServerAuthenticationPlugin.class,
                             pluginManager);
 
+            serverAuthenticatorPlugin.configure(configuration);
+
             serverAuthenticators.put(
                     protocolEntry.getKey(),
                     () -> serverAuthenticatorPlugin.createServerAuthenticator(configuration));

fluss-common/src/main/java/org/apache/fluss/security/auth/ServerAuthenticationPlugin.java

@@ -33,6 +33,12 @@
 @PublicEvolving
 public interface ServerAuthenticationPlugin extends AuthenticationPlugin {
 
+    /**
+     * Prepares shared server authentication configuration. Called once when the plugin is loaded,
+     * before creating connection-local authenticators.
+     */
+    default void configure(Configuration configuration) {}
+
     /**
      * Creates a server-side authenticator instance for this authentication protocol.
      *

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/SaslAuthenticatorFactory.java

@@ -31,6 +31,7 @@
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
+import java.util.function.BiFunction;
 import java.util.function.Function;
 
 /** Factory for creating connection-local SASL authenticators. */
@@ -44,7 +45,7 @@ public final class SaslAuthenticatorFactory {
                 PlainSaslServer.PLAIN_MECHANISM,
                 new AuthenticatorCreators(
                         configuration -> new SaslClientAuthenticator(configuration),
-                        configuration -> new PlainSaslServerAuthenticator(configuration)));
+                        PlainSaslServerAuthenticator::new));
         AUTHENTICATOR_CREATORS = Collections.unmodifiableMap(creators);
     }
 
@@ -65,13 +66,20 @@ public static ClientAuthenticator createClientAuthenticator(Configuration config
     /** Creates a server authenticator for the requested mechanism. */
     public static ServerAuthenticator createServerAuthenticator(
             String mechanism, Configuration configuration) {
+        return createServerAuthenticator(
+                mechanism, configuration, new SaslAuthorization(configuration));
+    }
+
+    /** Creates a server authenticator with shared, precompiled authorization rules. */
+    public static ServerAuthenticator createServerAuthenticator(
+            String mechanism, Configuration configuration, SaslAuthorization authorization) {
         String normalizedMechanism = mechanism.toUpperCase(Locale.ROOT);
         AuthenticatorCreators creators = AUTHENTICATOR_CREATORS.get(normalizedMechanism);
         if (creators == null) {
             throw new AuthenticationException(
                     "Unable to find a matching SASL mechanism for " + normalizedMechanism);
         }
-        return creators.serverCreator.apply(configuration);
+        return creators.serverCreator.apply(configuration, authorization);
     }
 
     /** Returns whether the client supports the requested mechanism. */
@@ -86,11 +94,12 @@ public static boolean supportsServerMechanism(String mechanism) {
 
     private static final class AuthenticatorCreators {
         private final Function<Configuration, ClientAuthenticator> clientCreator;
-        private final Function<Configuration, ServerAuthenticator> serverCreator;
+        private final BiFunction<Configuration, SaslAuthorization, ServerAuthenticator>
+                serverCreator;
 
         private AuthenticatorCreators(
                 Function<Configuration, ClientAuthenticator> clientCreator,
-                Function<Configuration, ServerAuthenticator> serverCreator) {
+                BiFunction<Configuration, SaslAuthorization, ServerAuthenticator> serverCreator) {
             this.clientCreator = clientCreator;
             this.serverCreator = serverCreator;
         }

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/SaslAuthorization.java

@@ -0,0 +1,146 @@
+/*
+ * 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.security.auth.sasl;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.AuthenticationException;
+import org.apache.fluss.exception.ConfigException;
+import org.apache.fluss.security.acl.FlussPrincipal;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+/** Immutable proxy grants for SASL authorization identities. */
+@Internal
+public final class SaslAuthorization {
+    private static final Logger LOG = LoggerFactory.getLogger(SaslAuthorization.class);
+    private static final String REGEX_PREFIX = "regex:";
+
+    private final Map<String, List<Pattern>> allowedTargetUsers;
+    private final Set<FlussPrincipal> superUsers;
+    private final boolean principalIgnoreCase;
+
+    /** Compiles and validates the configured proxy grants. */
+    public SaslAuthorization(Configuration configuration) {
+        this.allowedTargetUsers = parseProxyUsers(configuration);
+        this.superUsers =
+                configuration
+                        .getOptional(ConfigOptions.SUPER_USERS)
+                        .map(FlussPrincipal::parsePrincipals)
+                        .orElse(Collections.emptySet());
+        this.principalIgnoreCase =
+                configuration.get(ConfigOptions.SECURITY_ACL_PRINCIPAL_IGNORE_CASE);
+    }
+
+    /** Returns the effective identity after checking the authenticated account's proxy grants. */
+    public String authorize(String authenticatedUser, String requestedUser) {
+        if (requestedUser.isEmpty() || requestedUser.equals(authenticatedUser)) {
+            return authenticatedUser;
+        }
+        if (superUsers.stream()
+                .anyMatch(superUser -> superUser.matchesName(requestedUser, principalIgnoreCase))) {
+            throw new AuthenticationException(
+                    String.format(
+                            "Authentication failed: user '%s' is not allowed to act as '%s' because it is a super user",
+                            authenticatedUser, requestedUser));
+        }
+        if (!mayActAs(authenticatedUser, requestedUser)) {
+            throw new AuthenticationException(
+                    String.format(
+                            "Authentication failed: user '%s' is not allowed to act as '%s'",
+                            authenticatedUser, requestedUser));
+        }
+        LOG.info(
+                "User '{}' authenticated and is acting as '{}'",
+                escapeForLog(authenticatedUser),
+                escapeForLog(requestedUser));
+        return requestedUser;
+    }
+
+    private boolean mayActAs(String authenticatedUser, String requestedUser) {
+        List<Pattern> allowed = allowedTargetUsers.get(authenticatedUser);
+        return allowed != null
+                && allowed.stream().anyMatch(pattern -> pattern.matcher(requestedUser).matches());
+    }
+
+    private static String escapeForLog(String value) {
+        return value.replace("\r", "\\r").replace("\n", "\\n");
+    }
+
+    private static Map<String, List<Pattern>> parseProxyUsers(Configuration configuration) {
+        Map<String, List<Pattern>> allowedTargetUsers = new HashMap<>();
+        Map<String, String> proxyUsers =
+                configuration
+                        .getOptional(ConfigOptions.SERVER_SASL_PROXY_USERS)
+                        .orElse(Collections.emptyMap());
+        for (Map.Entry<String, String> entry : proxyUsers.entrySet()) {
+            List<Pattern> patterns = new ArrayList<>();
+            for (String target : splitOnSemicolon(entry.getValue())) {
+                patterns.add(compileGrant(entry.getKey(), target));
+            }
+            allowedTargetUsers.put(entry.getKey().trim(), Collections.unmodifiableList(patterns));
+        }
+        return Collections.unmodifiableMap(allowedTargetUsers);
+    }
+
+    private static Pattern compileGrant(String proxyUser, String grant) {
+        String regex;
+        if ("*".equals(grant)) {
+            regex = "(?s:.*)";
+        } else if (grant.startsWith(REGEX_PREFIX)) {
+            regex = grant.substring(REGEX_PREFIX.length());
+        } else {
+            regex = Pattern.quote(grant);
+        }
+        try {
+            return Pattern.compile(regex);
+        } catch (PatternSyntaxException e) {
+            throw new ConfigException(
+                    String.format(
+                            "Invalid rule '%s' for proxy user '%s' in '%s': %s",
+                            grant,
+                            proxyUser,
+                            ConfigOptions.SERVER_SASL_PROXY_USERS.key(),
+                            e.getDescription()),
+                    e);
+        }
+    }
+
+    private static Set<String> splitOnSemicolon(String value) {
+        Set<String> values = new HashSet<>();
+        for (String part : value.split(";")) {
+            String trimmed = part.trim();
+            if (!trimmed.isEmpty()) {
+                values.add(trimmed);
+            }
+        }
+        return values;
+    }
+}