[security] Support SASL/PLAIN proxy authorization

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

@@ -349,6 +349,56 @@ void testAlterDatabase() throws Exception {
         assertThat(rootAdmin.databaseExists("test-database2").get()).isTrue();
     }
 
+    @Test
+    void testImpersonationUsesTargetAcls() throws Exception {
+        String targetUser = "tenant_a_guest";
+        Configuration serviceConf = FLUSS_CLUSTER_EXTENSION.getClientConfig("CLIENT");
+        serviceConf.set(ConfigOptions.CLIENT_SECURITY_PROTOCOL, "SASL");
+        serviceConf.set(ConfigOptions.CLIENT_SASL_JAAS_USERNAME, "proxy");
+        serviceConf.set(ConfigOptions.CLIENT_SASL_JAAS_PASSWORD, "password3");
+        Configuration proxyConf = new Configuration(serviceConf);
+        proxyConf.set(ConfigOptions.CLIENT_SASL_AUTHORIZATION_ID, targetUser);
+        FlussPrincipal targetPrincipal = new FlussPrincipal(targetUser, "User");
+        FlussPrincipal servicePrincipal = new FlussPrincipal("proxy", "User");
+        try (Connection serviceConn = ConnectionFactory.createConnection(serviceConf);
+                Admin serviceAdmin = serviceConn.getAdmin();
+                Connection proxyConn = ConnectionFactory.createConnection(proxyConf);
+                Admin proxyAdmin = proxyConn.getAdmin()) {
+            List<AclBinding> serviceAcls =
+                    Collections.singletonList(
+                            new AclBinding(
+                                    Resource.table(DATA1_TABLE_PATH_PK),
+                                    new AccessControlEntry(
+                                            servicePrincipal,
+                                            WILD_CARD_HOST,
+                                            OperationType.DESCRIBE,
+                                            PermissionType.ALLOW)));
+            rootAdmin.createAcls(serviceAcls).all().get();
+            FLUSS_CLUSTER_EXTENSION.waitUntilAuthenticationSync(serviceAcls, true);
+            assertThat(serviceAdmin.getTableInfo(DATA1_TABLE_PATH_PK).get().getTablePath())
+                    .isEqualTo(DATA1_TABLE_PATH_PK);
+            assertThatThrownBy(() -> proxyAdmin.getTableInfo(DATA1_TABLE_PATH_PK).get())
+                    .cause()
+                    .isInstanceOf(AuthorizationException.class)
+                    .hasMessageContaining(
+                            "Principal " + targetPrincipal + " have no authorization");
+
+            List<AclBinding> aclBindings =
+                    Collections.singletonList(
+                            new AclBinding(
+                                    Resource.table(DATA1_TABLE_PATH_PK),
+                                    new AccessControlEntry(
+                                            targetPrincipal,
+                                            WILD_CARD_HOST,
+                                            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);
+        }
+    }
+
     @Test
     void testListDatabases() throws ExecutionException, InterruptedException {
         assertThat(guestAdmin.listDatabases().get())
@@ -1520,7 +1570,10 @@ 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.setString(
+                "security.sasl.listener.name.client.plain.proxy-users", "proxy:tenant_a_guest");
         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

@@ -1993,7 +1993,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 +2002,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.listener.name.<listener>.plain.proxy-users'.");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for Fluss Table
     // ------------------------------------------------------------------------

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

@@ -44,7 +44,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);
     }
 

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

@@ -18,13 +18,15 @@
 package org.apache.fluss.security.auth.sasl.authenticator;
 
 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.security.acl.FlussPrincipal;
 import org.apache.fluss.security.auth.ServerAuthenticator;
 import org.apache.fluss.security.auth.sasl.jaas.JaasContext;
 import org.apache.fluss.security.auth.sasl.jaas.LoginManager;
 import org.apache.fluss.security.auth.sasl.jaas.SaslServerFactory;
+import org.apache.fluss.security.auth.sasl.plain.PlainSaslProxyAuthorization;
 import org.apache.fluss.security.auth.sasl.plain.PlainSaslServer;
 
 import org.slf4j.Logger;
@@ -34,21 +36,22 @@
 import javax.security.sasl.SaslServer;
 
 import java.util.Locale;
-import java.util.Map;
+
+import static org.apache.fluss.config.ConfigBuilder.key;
 
 /** A connection-local SASL/PLAIN server authenticator. */
 @Internal
 public final class PlainSaslServerAuthenticator implements ServerAuthenticator {
     private static final Logger LOG = LoggerFactory.getLogger(PlainSaslServerAuthenticator.class);
     private static final String SERVER_AUTHENTICATOR_PREFIX = "security.sasl.";
 
-    private final Map<String, String> configs;
+    private final Configuration configuration;
     private SaslServer saslServer;
     private LoginManager loginManager;
 
-    /** Creates an authenticator using a snapshot of the supplied configuration. */
+    /** Creates an authenticator that reads credentials and proxy grants when initialized. */
     public PlainSaslServerAuthenticator(Configuration configuration) {
-        this.configs = configuration.toMap();
+        this.configuration = configuration;
     }
 
     @Override
@@ -68,10 +71,9 @@ public void initialize(AuthenticateContext context) {
             SaslServer newSaslServer =
                     SaslServerFactory.createSaslServer(
                             PlainSaslServer.PLAIN_MECHANISM,
-                            context.ipAddress(),
-                            configs,
-                            acquiredLoginManager,
-                            contextConfig.configurationEntries());
+                            contextConfig.configurationEntries(),
+                            PlainSaslProxyAuthorization.loadForListener(
+                                    configuration, context.listenerName()));
             loginManager = acquiredLoginManager;
             saslServer = newSaslServer;
         } catch (Exception e) {
@@ -128,21 +130,18 @@ private String findJaasConfig(String listenerName) {
                                 + JaasContext.SASL_JAAS_CONFIG,
                         listenerName.toLowerCase(Locale.ROOT),
                         PlainSaslServer.PLAIN_MECHANISM.toLowerCase(Locale.ROOT));
-        String dynamicJaasConfig = configs.get(listenerMechanismKey);
+        String dynamicJaasConfig =
+                configuration.get(key(listenerMechanismKey).stringType().noDefaultValue());
         if (dynamicJaasConfig != null && !dynamicJaasConfig.isEmpty()) {
             return dynamicJaasConfig;
         }
 
-        String globalMechanismKey =
-                SERVER_AUTHENTICATOR_PREFIX
-                        + PlainSaslServer.PLAIN_MECHANISM.toLowerCase(Locale.ROOT)
-                        + "."
-                        + JaasContext.SASL_JAAS_CONFIG;
+        String globalMechanismKey = ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG.key();
         LOG.debug(
                 "No listener-mechanism JAAS config found for key: '{}'. Falling back to mechanism-level config: '{}'",
                 listenerMechanismKey,
                 globalMechanismKey);
-        dynamicJaasConfig = configs.get(globalMechanismKey);
+        dynamicJaasConfig = configuration.get(ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG);
         if (dynamicJaasConfig == null || dynamicJaasConfig.isEmpty()) {
             LOG.warn(
                     "No mechanism-level JAAS config found for key: '{}'. Falling back to JVM option: -D{}",

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

@@ -17,6 +17,7 @@
 
 package org.apache.fluss.security.auth.sasl.jaas;
 
+import org.apache.fluss.security.auth.sasl.plain.PlainSaslProxyAuthorization;
 import org.apache.fluss.security.auth.sasl.plain.PlainSaslServer;
 import org.apache.fluss.security.auth.sasl.plain.PlainServerCallbackHandler;
 
@@ -27,7 +28,6 @@
 import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.sasl.Sasl;
 import javax.security.sasl.SaslClient;
-import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslServer;
 
 import java.security.PrivilegedActionException;
@@ -36,6 +36,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.fluss.config.ConfigOptions.CLIENT_SASL_AUTHORIZATION_ID;
+
 /* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache
  * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
  * additional information regarding copyright ownership. */
@@ -44,19 +46,16 @@
 public class SaslServerFactory {
     private static final Logger LOG = LoggerFactory.getLogger(SaslServerFactory.class);
 
+    /** Creates a PLAIN server using the listener's proxy authorization policy. */
     public static SaslServer createSaslServer(
             String mechanism,
-            String hostName,
-            Map<String, ?> props,
-            LoginManager loginManager,
-            List<AppConfigurationEntry> configurationEntries)
-            throws SaslException {
-
+            List<AppConfigurationEntry> configurationEntries,
+            PlainSaslProxyAuthorization authorization) {
         if (!PlainSaslServer.PLAIN_MECHANISM.equals(mechanism)) {
             throw new IllegalArgumentException("Unsupported mechanism: " + mechanism);
         }
 
-        AuthenticateCallbackHandler callbackHandler = new PlainServerCallbackHandler();
+        AuthenticateCallbackHandler callbackHandler = new PlainServerCallbackHandler(authorization);
         callbackHandler.configure(mechanism, configurationEntries);
         // Kafka and Fluss register PLAIN providers with the same JVM provider name. Construct
         // Fluss's server directly so callback types do not depend on provider registration order.
@@ -73,14 +72,18 @@ public static SaslClient createSaslClient(
                         () -> {
                             String[] mechs = {mechanism};
                             String serviceName = loginManager.serviceName();
+                            String authorizationId =
+                                    (String) props.get(CLIENT_SASL_AUTHORIZATION_ID.key());
                             LOG.debug(
                                     "Creating SaslClient: service={};mechs={}",
                                     serviceName,
                                     Arrays.toString(mechs));
 
                             return Sasl.createSaslClient(
                                     mechs,
-                                    null,
+                                    authorizationId == null || authorizationId.isEmpty()
+                                            ? null
+                                            : authorizationId,
                                     serviceName,
                                     hostAddress,
                                     props,