[server] Support group-based ACL authorization

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

@@ -21,6 +21,8 @@
 import org.apache.fluss.exception.AuthenticationException;
 import org.apache.fluss.security.acl.FlussPrincipal;
 
+import java.util.Collections;
+
 /** Authentication Plugin for PLAINTEXT which not need to do authentication. */
 public class PlainTextAuthenticationPlugin
         implements ServerAuthenticationPlugin, ClientAuthenticationPlugin {
@@ -73,8 +75,8 @@ public byte[] evaluateResponse(byte[] token) {
         }
 
         @Override
-        public FlussPrincipal createPrincipal() {
-            return FlussPrincipal.ANONYMOUS;
+        public Iterable<FlussPrincipal> createPrincipals() {
+            return Collections.singletonList(FlussPrincipal.ANONYMOUS);
         }
 
         @Override

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

@@ -91,10 +91,15 @@ default void initialize(AuthenticateContext context) {}
     boolean isCompleted();
 
     /**
-     * Create principal from authenticated token for later authorization.(this can only invoke if is
-     * complete).
+     * Creates the principals associated with the authenticated token for authorization. This method
+     * may only be invoked after authentication completes.
+     *
+     * <p>The iterable must be non-null and non-empty, with no {@code null} elements. All principals
+     * participate equally in ACL and superuser checks. The first principal identifies the client
+     * for logging, user metrics, and resource ownership; subsequent principals may represent groups
+     * or roles.
      */
-    FlussPrincipal createPrincipal();
+    Iterable<FlussPrincipal> createPrincipals();
 
     /** Close the authenticator. */
     default void close() throws IOException {}

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

@@ -33,6 +33,7 @@
 import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslServer;
 
+import java.util.Collections;
 import java.util.Locale;
 import java.util.Map;
 
@@ -99,8 +100,9 @@ public boolean isCompleted() {
     }
 
     @Override
-    public FlussPrincipal createPrincipal() {
-        return new FlussPrincipal(saslServer.getAuthorizationID(), "User");
+    public Iterable<FlussPrincipal> createPrincipals() {
+        return Collections.singletonList(
+                new FlussPrincipal(saslServer.getAuthorizationID(), "User"));
     }
 
     @Override

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

@@ -94,8 +94,8 @@ public boolean isCompleted() {
     }
 
     @Override
-    public FlussPrincipal createPrincipal() {
-        return delegate.createPrincipal();
+    public Iterable<FlussPrincipal> createPrincipals() {
+        return delegate.createPrincipals();
     }
 
     @Override

fluss-common/src/test/java/org/apache/fluss/security/auth/TestIdentifierAuthenticationPlugin.java

@@ -23,6 +23,8 @@
 
 import javax.annotation.Nullable;
 
+import java.util.Collections;
+
 /** A {@link AuthenticationPlugin} that is used to test authentication plugin identifier. */
 public class TestIdentifierAuthenticationPlugin
         implements ClientAuthenticationPlugin, ServerAuthenticationPlugin {
@@ -73,8 +75,8 @@ public byte[] evaluateResponse(byte[] token) throws AuthenticationException {
         }
 
         @Override
-        public FlussPrincipal createPrincipal() {
-            return null;
+        public Iterable<FlussPrincipal> createPrincipals() {
+            return Collections.singletonList(FlussPrincipal.ANONYMOUS);
         }
 
         @Override

fluss-kafka/src/main/java/org/apache/fluss/kafka/security/KafkaSaslConnection.java

@@ -197,7 +197,7 @@ public synchronized byte[] authenticate(byte[] token) {
         try {
             byte[] challenge = authenticator.evaluateResponse(checkNotNull(token));
             if (authenticator.isCompleted()) {
-                principal = checkNotNull(authenticator.createPrincipal());
+                principal = checkNotNull(authenticator.createPrincipals().iterator().next());
                 state = State.READY;
                 closeAuthenticator(authenticator);
                 authenticator = null;

fluss-kafka/src/test/java/org/apache/fluss/kafka/TestingSaslAuthenticator.java

@@ -21,6 +21,8 @@
 import org.apache.fluss.security.acl.FlussPrincipal;
 import org.apache.fluss.security.auth.ServerAuthenticator;
 
+import java.util.Collections;
+
 /** Controllable authenticator for failure cleanup and request identity tests. */
 public class TestingSaslAuthenticator implements ServerAuthenticator {
     private final FlussPrincipal principal;
@@ -69,8 +71,8 @@ public boolean isCompleted() {
     }
 
     @Override
-    public FlussPrincipal createPrincipal() {
-        return principal;
+    public Iterable<FlussPrincipal> createPrincipals() {
+        return Collections.singletonList(principal);
     }
 
     @Override

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequest.java

@@ -21,13 +21,15 @@
 import org.apache.fluss.rpc.protocol.ApiMethod;
 import org.apache.fluss.rpc.protocol.RequestType;
 import org.apache.fluss.security.acl.FlussPrincipal;
+import org.apache.fluss.shaded.guava32.com.google.common.collect.ImmutableList;
 import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf;
 import org.apache.fluss.shaded.netty4.io.netty.util.ReferenceCountUtil;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.InetAddress;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
@@ -44,7 +46,7 @@ public final class FlussRequest implements RpcRequest {
     private final ByteBuf buffer;
     private final String listenerName;
     private final boolean isInternal;
-    private final FlussPrincipal principal;
+    private final List<FlussPrincipal> principals;
     private final InetAddress address;
     private final CompletableFuture<ApiMessage> responseFuture;
 
@@ -54,6 +56,7 @@ public final class FlussRequest implements RpcRequest {
     private volatile long requestCompletedTimeMs;
     private volatile boolean cancelled = false;
 
+    /** Creates a request with an immutable snapshot of the authenticated principals. */
     public FlussRequest(
             short apiKey,
             short apiVersion,
@@ -63,7 +66,7 @@ public FlussRequest(
             ByteBuf buffer,
             String listenerName,
             boolean isInternal,
-            FlussPrincipal principal,
+            Iterable<FlussPrincipal> principals,
             InetAddress address,
             CompletableFuture<ApiMessage> responseFuture) {
         this.apiKey = apiKey;
@@ -74,7 +77,7 @@ public FlussRequest(
         this.buffer = checkNotNull(buffer);
         this.responseFuture = responseFuture;
         this.isInternal = isInternal;
-        this.principal = principal;
+        this.principals = ImmutableList.copyOf(principals);
         this.address = address;
         this.listenerName = listenerName;
         this.startTimeMs = System.currentTimeMillis();
@@ -159,8 +162,9 @@ public InetAddress getAddress() {
         return address;
     }
 
-    public FlussPrincipal getPrincipal() {
-        return principal;
+    /** Returns the authenticated principals associated with this request. */
+    public List<FlussPrincipal> getPrincipals() {
+        return principals;
     }
 
     public boolean isInternal() {

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java

@@ -62,7 +62,7 @@ public void processRequest(FlussRequest request) {
                             request.getListenerName(),
                             request.isInternal(),
                             request.getAddress(),
-                            request.getPrincipal()));
+                            request.getPrincipals()));
             // check if the coordinator server is the current leader if the API is a coordinator
             // TODO: we should only check coordinator APIs instead of all APIs
             if (isCoordinator && api.getApiKey() != ApiKeys.API_VERSIONS) {

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java

@@ -50,6 +50,7 @@
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.ArrayDeque;
+import java.util.Collections;
 import java.util.Deque;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -133,7 +134,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                             buffer,
                             listenerName,
                             isInternal,
-                            authenticator.isCompleted() ? authenticator.createPrincipal() : null,
+                            authenticator.isCompleted()
+                                    ? authenticator.createPrincipals()
+                                    : Collections.emptyList(),
                             ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress(),
                             future);
 

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/Session.java

@@ -18,29 +18,47 @@
 package org.apache.fluss.rpc.netty.server;
 
 import org.apache.fluss.security.acl.FlussPrincipal;
+import org.apache.fluss.shaded.guava32.com.google.common.collect.ImmutableList;
 
 import java.io.Serializable;
 import java.net.InetAddress;
+import java.util.Collections;
+import java.util.List;
 
 /** The connection session of a request. */
 public class Session implements Serializable {
     private final short apiVersion;
     private final String listenerName;
     private final boolean isInternal;
     private final InetAddress inetAddress;
-    private final FlussPrincipal principal;
+    private final List<FlussPrincipal> principals;
 
     public Session(
             short apiVersion,
             String listenerName,
             boolean isInternal,
             InetAddress inetAddress,
             FlussPrincipal principal) {
+        this(
+                apiVersion,
+                listenerName,
+                isInternal,
+                inetAddress,
+                Collections.singletonList(principal));
+    }
+
+    /** Creates a session with an immutable snapshot of the authenticated principals. */
+    public Session(
+            short apiVersion,
+            String listenerName,
+            boolean isInternal,
+            InetAddress inetAddress,
+            Iterable<FlussPrincipal> principals) {
         this.apiVersion = apiVersion;
         this.listenerName = listenerName;
         this.isInternal = isInternal;
         this.inetAddress = inetAddress;
-        this.principal = principal;
+        this.principals = ImmutableList.copyOf(principals);
     }
 
     public short getApiVersion() {
@@ -55,8 +73,14 @@ public InetAddress getInetAddress() {
         return inetAddress;
     }
 
+    /** Returns the client identity used for logging, user metrics, and resource ownership. */
     public FlussPrincipal getPrincipal() {
-        return principal;
+        return principals.isEmpty() ? null : principals.get(0);
+    }
+
+    /** Returns all authenticated principals used for authorization. */
+    public List<FlussPrincipal> getPrincipals() {
+        return principals;
     }
 
     public boolean isInternal() {

fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java

@@ -32,7 +32,9 @@
 import org.apache.fluss.rpc.netty.client.NettyClient;
 import org.apache.fluss.rpc.netty.server.NettyServer;
 import org.apache.fluss.rpc.netty.server.RequestsMetrics;
+import org.apache.fluss.rpc.netty.server.Session;
 import org.apache.fluss.rpc.protocol.ApiKeys;
+import org.apache.fluss.security.acl.FlussPrincipal;
 import org.apache.fluss.utils.NetUtils;
 
 import org.junit.jupiter.api.AfterEach;
@@ -52,6 +54,7 @@ public class AuthenticationTest {
     private NettyServer nettyServer;
     private ServerNode usernamePasswordServerNode;
     private ServerNode mutualAuthServerNode;
+    private TestingAuthenticateGatewayService service;
 
     @BeforeEach
     public void setup() throws Exception {
@@ -106,6 +109,12 @@ void testMutualAuthenticate() throws Exception {
         try (NettyClient nettyClient =
                 new NettyClient(clientConfig, TestingClientMetricGroup.newInstance())) {
             verifyGetTableNamesList(nettyClient, mutualAuthServerNode);
+            assertThat(service.getLastSession().getPrincipal()).isEqualTo(FlussPrincipal.ANONYMOUS);
+            assertThat(service.getLastSession().getPrincipals())
+                    .containsExactly(
+                            FlussPrincipal.ANONYMOUS,
+                            new FlussPrincipal("data-engineers", "Group"),
+                            new FlussPrincipal("table-reader", "Role"));
         }
 
         // test invalid challenge from server
@@ -248,7 +257,7 @@ private void buildNettyServer() throws Exception {
         // 3 worker threads is enough for this test
         configuration.setString(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS.key(), "3");
         MetricGroup metricGroup = NOPMetricsGroup.newInstance();
-        TestingAuthenticateGatewayService service = new TestingAuthenticateGatewayService();
+        service = new TestingAuthenticateGatewayService();
         try (NetUtils.Port availablePort1 = getAvailablePort();
                 NetUtils.Port availablePort2 = getAvailablePort()) {
             this.nettyServer =
@@ -277,10 +286,17 @@ private void buildNettyServer() throws Exception {
      * authentication.
      */
     public static class TestingAuthenticateGatewayService extends TestingTabletGatewayService {
+        private volatile Session lastSession;
+
         @Override
         public CompletableFuture<ListTablesResponse> listTables(ListTablesRequest request) {
+            lastSession = currentSession();
             return CompletableFuture.completedFuture(
                     new ListTablesResponse().addAllTableNames(Collections.singleton("test-table")));
         }
+
+        Session getLastSession() {
+            return lastSession;
+        }
     }
 }

fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/MutualAuthenticationPlugin.java

@@ -27,6 +27,7 @@
 import org.apache.fluss.security.auth.ServerAuthenticationPlugin;
 import org.apache.fluss.security.auth.ServerAuthenticator;
 
+import java.util.Arrays;
 import java.util.concurrent.ThreadLocalRandom;
 
 import static org.apache.fluss.config.ConfigBuilder.key;
@@ -190,8 +191,11 @@ public byte[] evaluateResponse(byte[] token) throws AuthenticationException {
         }
 
         @Override
-        public FlussPrincipal createPrincipal() {
-            return FlussPrincipal.ANONYMOUS;
+        public Iterable<FlussPrincipal> createPrincipals() {
+            return Arrays.asList(
+                    FlussPrincipal.ANONYMOUS,
+                    new FlussPrincipal("data-engineers", "Group"),
+                    new FlussPrincipal("table-reader", "Role"));
         }
 
         @Override