From 34008c40e29c73dccd0c3bbf51940f23e00b8021 Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:02:05 +0800 Subject: [PATCH 1/3] [server] Support group-based ACL authorization --- .../security/auth/ServerAuthenticator.java | 14 +++++ .../fluss/rpc/netty/server/FlussRequest.java | 39 ++++++++++++++ .../rpc/netty/server/FlussRequestHandler.java | 3 +- .../rpc/netty/server/NettyServerHandler.java | 4 ++ .../fluss/rpc/netty/server/Session.java | 22 ++++++++ .../authenticate/AuthenticationTest.java | 17 +++++- .../MutualAuthenticationPlugin.java | 9 ++++ .../server/authorizer/DefaultAuthorizer.java | 21 +++++++- .../authorizer/DefaultAuthorizerTest.java | 53 ++++++++++++++++++- website/docs/security/authorization.md | 2 +- website/docs/security/overview.md | 4 +- 11 files changed, 181 insertions(+), 7 deletions(-) diff --git a/fluss-common/src/main/java/org/apache/fluss/security/auth/ServerAuthenticator.java b/fluss-common/src/main/java/org/apache/fluss/security/auth/ServerAuthenticator.java index c3ad251dc09..b608c8c3a2f 100644 --- a/fluss-common/src/main/java/org/apache/fluss/security/auth/ServerAuthenticator.java +++ b/fluss-common/src/main/java/org/apache/fluss/security/auth/ServerAuthenticator.java @@ -23,6 +23,8 @@ import java.io.Closeable; import java.io.IOException; +import java.util.Collections; +import java.util.List; /** * Authenticator for server side. @@ -96,6 +98,18 @@ default void initialize(AuthenticateContext context) {} */ FlussPrincipal createPrincipal(); + /** + * Creates additional principals (such as groups or roles) associated with the authenticated + * identity. These principals are evaluated during ACL matching but are not considered for + * superuser checks. + * + *

Implementations must return a non-null list with no {@code null} elements; return an empty + * list when there are none. + */ + default List createAdditionalPrincipals() { + return Collections.emptyList(); + } + /** Close the authenticator. */ default void close() throws IOException {} diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequest.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequest.java index adc67298e9d..dbaa6fffd46 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequest.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequest.java @@ -28,6 +28,9 @@ import org.slf4j.LoggerFactory; import java.net.InetAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.concurrent.CompletableFuture; import static org.apache.fluss.utils.Preconditions.checkNotNull; @@ -45,6 +48,7 @@ public final class FlussRequest implements RpcRequest { private final String listenerName; private final boolean isInternal; private final FlussPrincipal principal; + private final List additionalPrincipals; private final InetAddress address; private final CompletableFuture responseFuture; @@ -66,6 +70,35 @@ public FlussRequest( FlussPrincipal principal, InetAddress address, CompletableFuture responseFuture) { + this( + apiKey, + apiVersion, + requestId, + apiMethod, + message, + buffer, + listenerName, + isInternal, + principal, + Collections.emptyList(), + address, + responseFuture); + } + + /** Creates a request with the primary and additional authenticated principals. */ + public FlussRequest( + short apiKey, + short apiVersion, + int requestId, + ApiMethod apiMethod, + ApiMessage message, + ByteBuf buffer, + String listenerName, + boolean isInternal, + FlussPrincipal principal, + List additionalPrincipals, + InetAddress address, + CompletableFuture responseFuture) { this.apiKey = apiKey; this.apiVersion = apiVersion; this.requestId = requestId; @@ -75,6 +108,8 @@ public FlussRequest( this.responseFuture = responseFuture; this.isInternal = isInternal; this.principal = principal; + this.additionalPrincipals = + Collections.unmodifiableList(new ArrayList<>(additionalPrincipals)); this.address = address; this.listenerName = listenerName; this.startTimeMs = System.currentTimeMillis(); @@ -163,6 +198,10 @@ public FlussPrincipal getPrincipal() { return principal; } + public List getAdditionalPrincipals() { + return additionalPrincipals; + } + public boolean isInternal() { return isInternal; } diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java index 2ff5e0691bb..57cda30d0a4 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java @@ -62,7 +62,8 @@ public void processRequest(FlussRequest request) { request.getListenerName(), request.isInternal(), request.getAddress(), - request.getPrincipal())); + request.getPrincipal(), + request.getAdditionalPrincipals())); // 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) { diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java index 3d22c35238e..5372400749b 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java @@ -48,6 +48,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; @@ -130,6 +131,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception listenerName, isInternal, authenticator.isCompleted() ? authenticator.createPrincipal() : null, + authenticator.isCompleted() + ? authenticator.createAdditionalPrincipals() + : Collections.emptyList(), ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress(), future); diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/Session.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/Session.java index 178d797359e..66bbdee176e 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/Session.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/Session.java @@ -21,6 +21,9 @@ import java.io.Serializable; import java.net.InetAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; /** The connection session of a request. */ public class Session implements Serializable { @@ -29,6 +32,7 @@ public class Session implements Serializable { private final boolean isInternal; private final InetAddress inetAddress; private final FlussPrincipal principal; + private final List additionalPrincipals; public Session( short apiVersion, @@ -36,11 +40,24 @@ public Session( boolean isInternal, InetAddress inetAddress, FlussPrincipal principal) { + this(apiVersion, listenerName, isInternal, inetAddress, principal, Collections.emptyList()); + } + + /** Creates a session with the primary and additional authenticated principals. */ + public Session( + short apiVersion, + String listenerName, + boolean isInternal, + InetAddress inetAddress, + FlussPrincipal principal, + List additionalPrincipals) { this.apiVersion = apiVersion; this.listenerName = listenerName; this.isInternal = isInternal; this.inetAddress = inetAddress; this.principal = principal; + this.additionalPrincipals = + Collections.unmodifiableList(new ArrayList<>(additionalPrincipals)); } public short getApiVersion() { @@ -59,6 +76,11 @@ public FlussPrincipal getPrincipal() { return principal; } + /** Returns additional principals associated with the authenticated identity. */ + public List getAdditionalPrincipals() { + return additionalPrincipals; + } + public boolean isInternal() { return isInternal; } diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java index 9b98d735a26..95f3e478ef8 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java +++ b/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,11 @@ 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().getAdditionalPrincipals()) + .containsExactly( + new FlussPrincipal("data-engineers", "Group"), + new FlussPrincipal("table-reader", "Role")); } // test invalid challenge from server @@ -248,7 +256,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 +285,17 @@ private void buildNettyServer() throws Exception { * authentication. */ public static class TestingAuthenticateGatewayService extends TestingTabletGatewayService { + private volatile Session lastSession; + @Override public CompletableFuture listTables(ListTablesRequest request) { + lastSession = currentSession(); return CompletableFuture.completedFuture( new ListTablesResponse().addAllTableNames(Collections.singleton("test-table"))); } + + Session getLastSession() { + return lastSession; + } } } diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/MutualAuthenticationPlugin.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/MutualAuthenticationPlugin.java index 0f380b700d8..07c2f4d4ef0 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/MutualAuthenticationPlugin.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/MutualAuthenticationPlugin.java @@ -27,6 +27,8 @@ import org.apache.fluss.security.auth.ServerAuthenticationPlugin; import org.apache.fluss.security.auth.ServerAuthenticator; +import java.util.Arrays; +import java.util.List; import java.util.concurrent.ThreadLocalRandom; import static org.apache.fluss.config.ConfigBuilder.key; @@ -194,6 +196,13 @@ public FlussPrincipal createPrincipal() { return FlussPrincipal.ANONYMOUS; } + @Override + public List createAdditionalPrincipals() { + return Arrays.asList( + new FlussPrincipal("data-engineers", "Group"), + new FlussPrincipal("table-reader", "Role")); + } + @Override public boolean isCompleted() { return status == Status.COMPLETED; 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..9f4a4c0e657 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 @@ -174,6 +174,7 @@ public boolean authorizeAction(Session session, Action action) { || aclsAllowAccess( action.getResource(), principal, + session.getAdditionalPrincipals(), action.getOperation(), session.getInetAddress().getHostAddress()); } @@ -481,9 +482,27 @@ private void updateAclChangedFlag(Resource resource) { @VisibleForTesting public boolean aclsAllowAccess( Resource resource, FlussPrincipal principal, OperationType operation, String host) { + return aclsAllowAccess(resource, principal, Collections.emptyList(), operation, host); + } + + private boolean aclsAllowAccess( + Resource resource, + FlussPrincipal principal, + List additionalPrincipals, + OperationType operation, + String host) { Set accessControlEntries = matchingAcls(resource); return isEmptyAclAndAuthorized(resource, accessControlEntries) - || allowAclExists(resource, principal, operation, host, accessControlEntries); + || allowAclExists(resource, principal, operation, host, accessControlEntries) + || additionalPrincipals.stream() + .anyMatch( + additionalPrincipal -> + allowAclExists( + resource, + additionalPrincipal, + operation, + host, + accessControlEntries)); } private boolean isEmptyAclAndAuthorized(Resource resource, Collection acls) { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java index a8ba299c020..2a6b2acf5a3 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java @@ -205,6 +205,47 @@ void testSimpleAuthorizations() throws Exception { assertThat(authorizer.authorizeActions(session, actions)).containsExactly(false, true); } + @Test + void testAuthorizationWithAdditionalPrincipals() throws Exception { + Resource resource = Resource.table("database1", "table1"); + FlussPrincipal matchingGroup = new FlussPrincipal("data-engineers", "GROUP"); + Session session = + createSession( + "USER", + "alice", + "192.168.1.1", + Arrays.asList(new FlussPrincipal("fluss-readers", "GROUP"), matchingGroup)); + + authorizer.addAcls( + createRootUserSession(), + Collections.singletonList( + new AclBinding( + resource, + new AccessControlEntry( + matchingGroup, + "192.168.1.1", + READ, + PermissionType.ALLOW)))); + + assertThat(authorizer.isAuthorized(session, READ, resource)).isTrue(); + } + + @Test + void testAdditionalPrincipalDoesNotGrantSuperUser() throws Exception { + Resource resource = Resource.database("database1"); + Session session = + createSession( + "USER", + "alice", + "192.168.1.1", + Collections.singletonList(new FlussPrincipal(ROOT_USER, "USER"))); + authorizer.addAcls( + createRootUserSession(), + Collections.singletonList(createAclBinding(resource, "another-user", "*", READ))); + + assertThat(authorizer.isAuthorized(session, READ, resource)).isFalse(); + } + @Test void testAclModificationRequiresAllPermission() throws Exception { Resource resource = Resource.database("database1"); @@ -748,12 +789,22 @@ private Session createSession(String username, String host) throws Exception { } private Session createSession(String userType, String username, String host) throws Exception { + return createSession(userType, username, host, Collections.emptyList()); + } + + private Session createSession( + String userType, + String username, + String host, + List additionalPrincipals) + throws Exception { return new Session( (byte) 1, "FLUSS", false, InetAddress.getByName(host), - new FlussPrincipal(username, userType)); + new FlussPrincipal(username, userType), + additionalPrincipals); } private AclBinding createAclBinding( diff --git a/website/docs/security/authorization.md b/website/docs/security/authorization.md index 4cd05d70799..ac2e52f1fc8 100644 --- a/website/docs/security/authorization.md +++ b/website/docs/security/authorization.md @@ -66,7 +66,7 @@ Fluss implements a permission inheritance model, where certain operations imply The FlussPrincipal is a core concept in the Fluss security architecture. It represents the identity of an authenticated entity (such as a user or service) and serves as the central bridge between authentication and authorization. Once a client successfully authenticates via a supported mechanism (e.g., SASL/PLAIN, Kerberos), a FlussPrincipal is created to represent that client's identity. This principal is then used throughout the system for access control decisions, linking who the user is with what they are allowed to do. -The principal type indicates the category of the principal (e. g., "User", "Group", "Role"), while the name identifies the specific entity within that category. By default, the simple authorizer uses "User" as the principal type, but custom authorizers can extend this to support role-based or group-based access control lists (ACLs). +The principal type indicates the category of the principal (e. g., "User", "Group", "Role"), while the name identifies the specific entity within that category. An authenticated session has one primary principal and may have additional principals such as groups or roles. The default authorizer evaluates ACLs against the primary and additional principals, while superuser checks use only the primary principal. Example usage: * `new FlussPrincipal("admin", "User")` – A standard user principal. * `new FlussPrincipal("admins", "Group")` – A group-based principal for authorization. diff --git a/website/docs/security/overview.md b/website/docs/security/overview.md index aecbf87a51d..554f72d6f7e 100644 --- a/website/docs/security/overview.md +++ b/website/docs/security/overview.md @@ -77,9 +77,9 @@ For example: * User `guest` might only be allowed to consume messages from a specific table. ### Fluss Principal – The Bridge Between Authentication and Authorization -Once a client successfully authenticates, Fluss creates a **Fluss Principal**, which represents the authenticated identity. This **Fluss Principal** is used throughout the system during authorization checks. +Once a client successfully authenticates, Fluss creates a primary **Fluss Principal**, which represents the authenticated identity. Authentication may also provide additional principals such as groups or roles. -The principal type indicates the category of the principal (e.g., "User", "Group", "Role"), while the name identifies the specific entity within that category. By default, the simple authorizer uses "User" as the principal type, but custom authorizers can extend this to support role-based or group-based access control lists (ACLs). +The principal type indicates the category of the principal (e.g., "User", "Group", "Role"), while the name identifies the specific entity within that category. The default authorizer evaluates ACLs against the primary and additional principals, while superuser checks use only the primary principal. Example usage: * `new FlussPrincipal("admin", "User")` – A standard user principal. * `new FlussPrincipal("admins", "Group")` – A group-based principal for authorization. From 91f01868f9db2a2b61ca276ba8e0c93ffeb6a46c Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Thu, 5 Nov 2026 10:47:02 +0800 Subject: [PATCH 2/3] [security] Use a unified collection of authenticated principals --- .../auth/PlainTextAuthenticationPlugin.java | 6 +- .../security/auth/ServerAuthenticator.java | 23 ++---- .../SaslServerAuthenticator.java | 6 +- .../TestIdentifierAuthenticationPlugin.java | 6 +- .../kafka/security/KafkaSaslConnection.java | 2 +- .../fluss/kafka/TestingSaslAuthenticator.java | 6 +- .../fluss/rpc/netty/server/FlussRequest.java | 51 +++---------- .../rpc/netty/server/FlussRequestHandler.java | 3 +- .../rpc/netty/server/NettyServerHandler.java | 3 +- .../fluss/rpc/netty/server/Session.java | 30 ++++---- .../authenticate/AuthenticationTest.java | 3 +- .../MutualAuthenticationPlugin.java | 9 +-- .../fluss/rpc/netty/server/SessionTest.java | 71 +++++++++++++++++++ .../rpc/protocol/RequestChannelTest.java | 9 +-- .../server/authorizer/DefaultAuthorizer.java | 35 ++++----- .../authorizer/DefaultAuthorizerTest.java | 68 ++++++++++++------ website/docs/security/authorization.md | 10 ++- website/docs/security/overview.md | 4 +- 18 files changed, 200 insertions(+), 145 deletions(-) create mode 100644 fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/SessionTest.java diff --git a/fluss-common/src/main/java/org/apache/fluss/security/auth/PlainTextAuthenticationPlugin.java b/fluss-common/src/main/java/org/apache/fluss/security/auth/PlainTextAuthenticationPlugin.java index 7b9e0911654..39524f08184 100644 --- a/fluss-common/src/main/java/org/apache/fluss/security/auth/PlainTextAuthenticationPlugin.java +++ b/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 createPrincipals() { + return Collections.singletonList(FlussPrincipal.ANONYMOUS); } @Override diff --git a/fluss-common/src/main/java/org/apache/fluss/security/auth/ServerAuthenticator.java b/fluss-common/src/main/java/org/apache/fluss/security/auth/ServerAuthenticator.java index b608c8c3a2f..84763bd65a9 100644 --- a/fluss-common/src/main/java/org/apache/fluss/security/auth/ServerAuthenticator.java +++ b/fluss-common/src/main/java/org/apache/fluss/security/auth/ServerAuthenticator.java @@ -23,8 +23,6 @@ import java.io.Closeable; import java.io.IOException; -import java.util.Collections; -import java.util.List; /** * Authenticator for server side. @@ -93,22 +91,15 @@ default void initialize(AuthenticateContext context) {} boolean isCompleted(); /** - * Create principal from authenticated token for later authorization.(this can only invoke if is - * complete). - */ - FlussPrincipal createPrincipal(); - - /** - * Creates additional principals (such as groups or roles) associated with the authenticated - * identity. These principals are evaluated during ACL matching but are not considered for - * superuser checks. + * Creates the principals associated with the authenticated token for authorization. This method + * may only be invoked after authentication completes. * - *

Implementations must return a non-null list with no {@code null} elements; return an empty - * list when there are none. + *

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. */ - default List createAdditionalPrincipals() { - return Collections.emptyList(); - } + Iterable createPrincipals(); /** Close the authenticator. */ default void close() throws IOException {} diff --git a/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslServerAuthenticator.java b/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslServerAuthenticator.java index 8225010c60f..08fcb59c0fb 100644 --- a/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslServerAuthenticator.java +++ b/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslServerAuthenticator.java @@ -30,6 +30,7 @@ import javax.security.sasl.SaslException; import javax.security.sasl.SaslServer; +import java.util.Collections; import java.util.List; import java.util.Locale; import java.util.Map; @@ -155,8 +156,9 @@ public boolean isCompleted() { } @Override - public FlussPrincipal createPrincipal() { - return new FlussPrincipal(saslServer.getAuthorizationID(), "User"); + public Iterable createPrincipals() { + return Collections.singletonList( + new FlussPrincipal(saslServer.getAuthorizationID(), "User")); } @Override diff --git a/fluss-common/src/test/java/org/apache/fluss/security/auth/TestIdentifierAuthenticationPlugin.java b/fluss-common/src/test/java/org/apache/fluss/security/auth/TestIdentifierAuthenticationPlugin.java index 91e0c998298..46cd8527a1c 100644 --- a/fluss-common/src/test/java/org/apache/fluss/security/auth/TestIdentifierAuthenticationPlugin.java +++ b/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 createPrincipals() { + return Collections.singletonList(FlussPrincipal.ANONYMOUS); } @Override diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/security/KafkaSaslConnection.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/security/KafkaSaslConnection.java index d93c5f92cd5..40edc9d4b83 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/security/KafkaSaslConnection.java +++ b/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; diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/TestingSaslAuthenticator.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/TestingSaslAuthenticator.java index f173bea74af..5db416d943e 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/TestingSaslAuthenticator.java +++ b/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 createPrincipals() { + return Collections.singletonList(principal); } @Override diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequest.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequest.java index dbaa6fffd46..c698fc0ce66 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequest.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequest.java @@ -21,6 +21,7 @@ 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; @@ -28,8 +29,6 @@ import org.slf4j.LoggerFactory; import java.net.InetAddress; -import java.util.ArrayList; -import java.util.Collections; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -47,8 +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 additionalPrincipals; + private final List principals; private final InetAddress address; private final CompletableFuture responseFuture; @@ -58,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, @@ -67,36 +66,7 @@ public FlussRequest( ByteBuf buffer, String listenerName, boolean isInternal, - FlussPrincipal principal, - InetAddress address, - CompletableFuture responseFuture) { - this( - apiKey, - apiVersion, - requestId, - apiMethod, - message, - buffer, - listenerName, - isInternal, - principal, - Collections.emptyList(), - address, - responseFuture); - } - - /** Creates a request with the primary and additional authenticated principals. */ - public FlussRequest( - short apiKey, - short apiVersion, - int requestId, - ApiMethod apiMethod, - ApiMessage message, - ByteBuf buffer, - String listenerName, - boolean isInternal, - FlussPrincipal principal, - List additionalPrincipals, + Iterable principals, InetAddress address, CompletableFuture responseFuture) { this.apiKey = apiKey; @@ -107,9 +77,7 @@ public FlussRequest( this.buffer = checkNotNull(buffer); this.responseFuture = responseFuture; this.isInternal = isInternal; - this.principal = principal; - this.additionalPrincipals = - Collections.unmodifiableList(new ArrayList<>(additionalPrincipals)); + this.principals = ImmutableList.copyOf(principals); this.address = address; this.listenerName = listenerName; this.startTimeMs = System.currentTimeMillis(); @@ -194,12 +162,9 @@ public InetAddress getAddress() { return address; } - public FlussPrincipal getPrincipal() { - return principal; - } - - public List getAdditionalPrincipals() { - return additionalPrincipals; + /** Returns the authenticated principals associated with this request. */ + public List getPrincipals() { + return principals; } public boolean isInternal() { diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java index 57cda30d0a4..148c050d297 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java @@ -62,8 +62,7 @@ public void processRequest(FlussRequest request) { request.getListenerName(), request.isInternal(), request.getAddress(), - request.getPrincipal(), - request.getAdditionalPrincipals())); + 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) { diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java index a8e398fec4a..b90a0644f51 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java @@ -134,9 +134,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception buffer, listenerName, isInternal, - authenticator.isCompleted() ? authenticator.createPrincipal() : null, authenticator.isCompleted() - ? authenticator.createAdditionalPrincipals() + ? authenticator.createPrincipals() : Collections.emptyList(), ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress(), future); diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/Session.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/Session.java index 66bbdee176e..550c8c2119e 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/Session.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/Session.java @@ -18,10 +18,10 @@ 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.ArrayList; import java.util.Collections; import java.util.List; @@ -31,8 +31,7 @@ public class Session implements Serializable { private final String listenerName; private final boolean isInternal; private final InetAddress inetAddress; - private final FlussPrincipal principal; - private final List additionalPrincipals; + private final List principals; public Session( short apiVersion, @@ -40,24 +39,26 @@ public Session( boolean isInternal, InetAddress inetAddress, FlussPrincipal principal) { - this(apiVersion, listenerName, isInternal, inetAddress, principal, Collections.emptyList()); + this( + apiVersion, + listenerName, + isInternal, + inetAddress, + Collections.singletonList(principal)); } - /** Creates a session with the primary and additional authenticated principals. */ + /** Creates a session with an immutable snapshot of the authenticated principals. */ public Session( short apiVersion, String listenerName, boolean isInternal, InetAddress inetAddress, - FlussPrincipal principal, - List additionalPrincipals) { + Iterable principals) { this.apiVersion = apiVersion; this.listenerName = listenerName; this.isInternal = isInternal; this.inetAddress = inetAddress; - this.principal = principal; - this.additionalPrincipals = - Collections.unmodifiableList(new ArrayList<>(additionalPrincipals)); + this.principals = ImmutableList.copyOf(principals); } public short getApiVersion() { @@ -72,13 +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 additional principals associated with the authenticated identity. */ - public List getAdditionalPrincipals() { - return additionalPrincipals; + /** Returns all authenticated principals used for authorization. */ + public List getPrincipals() { + return principals; } public boolean isInternal() { diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java index 95f3e478ef8..f0c8b801f5e 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java @@ -110,8 +110,9 @@ void testMutualAuthenticate() throws Exception { new NettyClient(clientConfig, TestingClientMetricGroup.newInstance())) { verifyGetTableNamesList(nettyClient, mutualAuthServerNode); assertThat(service.getLastSession().getPrincipal()).isEqualTo(FlussPrincipal.ANONYMOUS); - assertThat(service.getLastSession().getAdditionalPrincipals()) + assertThat(service.getLastSession().getPrincipals()) .containsExactly( + FlussPrincipal.ANONYMOUS, new FlussPrincipal("data-engineers", "Group"), new FlussPrincipal("table-reader", "Role")); } diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/MutualAuthenticationPlugin.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/MutualAuthenticationPlugin.java index 07c2f4d4ef0..da07c068e6e 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/MutualAuthenticationPlugin.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/MutualAuthenticationPlugin.java @@ -28,7 +28,6 @@ import org.apache.fluss.security.auth.ServerAuthenticator; import java.util.Arrays; -import java.util.List; import java.util.concurrent.ThreadLocalRandom; import static org.apache.fluss.config.ConfigBuilder.key; @@ -192,13 +191,9 @@ public byte[] evaluateResponse(byte[] token) throws AuthenticationException { } @Override - public FlussPrincipal createPrincipal() { - return FlussPrincipal.ANONYMOUS; - } - - @Override - public List createAdditionalPrincipals() { + public Iterable createPrincipals() { return Arrays.asList( + FlussPrincipal.ANONYMOUS, new FlussPrincipal("data-engineers", "Group"), new FlussPrincipal("table-reader", "Role")); } diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/SessionTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/SessionTest.java new file mode 100644 index 00000000000..7245ab0f110 --- /dev/null +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/SessionTest.java @@ -0,0 +1,71 @@ +/* + * 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.security.acl.FlussPrincipal; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for the authenticated principals in a {@link Session}. */ +class SessionTest { + + @Test + void testSnapshotsPrincipalsFromIterable() { + FlussPrincipal user = new FlussPrincipal("alice", "User"); + FlussPrincipal group = new FlussPrincipal("readers", "Group"); + List principals = new ArrayList<>(Arrays.asList(user, group)); + Iterable iterable = principals::iterator; + Session session = new Session((short) 0, "CLIENT", false, null, iterable); + + principals.clear(); + + assertThat(session.getPrincipal()).isEqualTo(user); + assertThat(session.getPrincipals()).containsExactly(user, group); + assertThatThrownBy(() -> session.getPrincipals().clear()) + .isInstanceOf(UnsupportedOperationException.class); + } + + @Test + void testRejectsNullPrincipals() { + assertThatThrownBy( + () -> + new Session( + (short) 0, + "CLIENT", + false, + null, + Arrays.asList(FlussPrincipal.ANONYMOUS, null))) + .isInstanceOf(NullPointerException.class); + } + + @Test + void testUnauthenticatedSessionHasNoPrincipals() { + Session session = new Session((short) 0, "CLIENT", false, null, Collections.emptyList()); + + assertThat(session.getPrincipals()).isEmpty(); + assertThat(session.getPrincipal()).isNull(); + } +} diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java index b433d008249..8adafd106f0 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java @@ -49,6 +49,7 @@ import java.net.InetSocketAddress; import java.net.SocketAddress; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -80,7 +81,7 @@ void testRequestsFIFO() throws Exception { new EmptyByteBuf(new UnpooledByteBufAllocator(true, true)), "FLUSS", true, - null, + Collections.emptyList(), null, new CompletableFuture<>()); channel.putRequest(rpcRequest); @@ -103,7 +104,7 @@ void testRequestsFIFO() throws Exception { new EmptyByteBuf(new UnpooledByteBufAllocator(true, true)), "FLUSS", true, - null, + Collections.emptyList(), null, new CompletableFuture<>()); RpcRequest rpcRequest2 = @@ -116,7 +117,7 @@ void testRequestsFIFO() throws Exception { new EmptyByteBuf(new UnpooledByteBufAllocator(true, true)), "FLUSS", true, - null, + Collections.emptyList(), null, new CompletableFuture<>()); channel.putRequest(rpcRequest1); @@ -305,7 +306,7 @@ private RpcRequest createTestRequest(int id) { new EmptyByteBuf(new UnpooledByteBufAllocator(true, true)), "FLUSS", true, - null, + Collections.emptyList(), null, new CompletableFuture<>()); } 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 10ee83fce5b..262cf27a8ce 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 @@ -173,21 +173,21 @@ public void close() { @Override public boolean authorizeAction(Session session, Action action) { - FlussPrincipal principal = session.getPrincipal(); - return isSuperUser(principal) + return isSuperUser(session.getPrincipals()) || aclsAllowAccess( action.getResource(), - principal, - session.getAdditionalPrincipals(), + session.getPrincipals(), action.getOperation(), session.getInetAddress().getHostAddress(), matchingAcls(action.getResource())); } - private boolean isSuperUser(FlussPrincipal principal) { - for (FlussPrincipal superUser : superUsers) { - if (superUser.matches(principal, principalIgnoreCase)) { - return true; + private boolean isSuperUser(List principals) { + for (FlussPrincipal principal : principals) { + for (FlussPrincipal superUser : superUsers) { + if (superUser.matches(principal, principalIgnoreCase)) { + return true; + } } } return false; @@ -346,13 +346,11 @@ private boolean isAuthorized( OperationType operationType, Resource resource, Map aclCacheSnapshot) { - FlussPrincipal principal = session.getPrincipal(); return session.isInternal() - || isSuperUser(principal) + || isSuperUser(session.getPrincipals()) || aclsAllowAccess( resource, - principal, - session.getAdditionalPrincipals(), + session.getPrincipals(), operationType, session.getInetAddress().getHostAddress(), matchingAcls(resource, aclCacheSnapshot)); @@ -539,8 +537,7 @@ public boolean aclsAllowAccess( Resource resource, FlussPrincipal principal, OperationType operation, String host) { return aclsAllowAccess( resource, - principal, - Collections.emptyList(), + Collections.singletonList(principal), operation, host, matchingAcls(resource)); @@ -548,19 +545,17 @@ public boolean aclsAllowAccess( private boolean aclsAllowAccess( Resource resource, - FlussPrincipal principal, - List additionalPrincipals, + List principals, OperationType operation, String host, Set accessControlEntries) { return isEmptyAclAndAuthorized(resource, accessControlEntries) - || allowAclExists(resource, principal, operation, host, accessControlEntries) - || additionalPrincipals.stream() + || principals.stream() .anyMatch( - additionalPrincipal -> + principal -> allowAclExists( resource, - additionalPrincipal, + principal, operation, host, accessControlEntries)); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java index 9701a8b235e..9c15dc8f23f 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java @@ -207,15 +207,16 @@ void testSimpleAuthorizations() throws Exception { } @Test - void testAuthorizationWithAdditionalPrincipals() throws Exception { + void testAuthorizationWithMultiplePrincipals() throws Exception { Resource resource = Resource.table("database1", "table1"); FlussPrincipal matchingGroup = new FlussPrincipal("data-engineers", "GROUP"); Session session = createSession( - "USER", - "alice", "192.168.1.1", - Arrays.asList(new FlussPrincipal("fluss-readers", "GROUP"), matchingGroup)); + Arrays.asList( + new FlussPrincipal("alice", "USER"), + new FlussPrincipal("fluss-readers", "GROUP"), + matchingGroup)); AclBinding groupAcl = new AclBinding( @@ -226,22 +227,54 @@ void testAuthorizationWithAdditionalPrincipals() throws Exception { assertThat(authorizer.isAuthorized(session, READ, resource)).isTrue(); assertThat(authorizer.listAcls(session, AclBindingFilter.ANY)).containsExactly(groupAcl); + Session reversedSession = + createSession( + "192.168.1.1", + Arrays.asList(matchingGroup, new FlussPrincipal("alice", "USER"))); + assertThat(authorizer.isAuthorized(reversedSession, READ, resource)).isTrue(); + assertThat(authorizer.listAcls(reversedSession, AclBindingFilter.ANY)) + .containsExactly(groupAcl); + + Session unrelatedSession = + createSession( + "192.168.1.1", + Arrays.asList( + new FlussPrincipal("alice", "USER"), + new FlussPrincipal("data-engineers", "USER"))); + assertThat(authorizer.isAuthorized(unrelatedSession, READ, resource)).isFalse(); } @Test - void testAdditionalPrincipalDoesNotGrantSuperUser() throws Exception { + void testAnyPrincipalCanGrantSuperUser() throws Exception { Resource resource = Resource.database("database1"); Session session = createSession( - "USER", - "alice", "192.168.1.1", - Collections.singletonList(new FlussPrincipal(ROOT_USER, "USER"))); + Arrays.asList( + new FlussPrincipal("alice", "USER"), + new FlussPrincipal(ROOT_USER, "USER"))); authorizer.addAcls( createRootUserSession(), Collections.singletonList(createAclBinding(resource, "another-user", "*", READ))); - assertThat(authorizer.isAuthorized(session, READ, resource)).isFalse(); + assertThat(authorizer.isAuthorized(session, READ, resource)).isTrue(); + assertThat(authorizer.listAcls(session, AclBindingFilter.ANY)).hasSize(1); + Session reversedSession = + createSession( + "192.168.1.1", + Arrays.asList( + new FlussPrincipal(ROOT_USER, "USER"), + new FlussPrincipal("alice", "USER"))); + assertThat(authorizer.isAuthorized(reversedSession, READ, resource)).isTrue(); + assertThat(authorizer.listAcls(reversedSession, AclBindingFilter.ANY)).hasSize(1); + + Session differentTypeSession = + createSession( + "192.168.1.1", + Arrays.asList( + new FlussPrincipal("alice", "USER"), + new FlussPrincipal(ROOT_USER, "GROUP"))); + assertThat(authorizer.isAuthorized(differentTypeSession, READ, resource)).isFalse(); } @Test @@ -866,22 +899,13 @@ private Session createSession(String username, String host) throws Exception { } private Session createSession(String userType, String username, String host) throws Exception { - return createSession(userType, username, host, Collections.emptyList()); + return createSession( + host, Collections.singletonList(new FlussPrincipal(username, userType))); } - private Session createSession( - String userType, - String username, - String host, - List additionalPrincipals) + private Session createSession(String host, Iterable principals) throws Exception { - return new Session( - (byte) 1, - "FLUSS", - false, - InetAddress.getByName(host), - new FlussPrincipal(username, userType), - additionalPrincipals); + return new Session((byte) 1, "FLUSS", false, InetAddress.getByName(host), principals); } private AclBinding createAclBinding( diff --git a/website/docs/security/authorization.md b/website/docs/security/authorization.md index cca77ccaef1..8a1a545c370 100644 --- a/website/docs/security/authorization.md +++ b/website/docs/security/authorization.md @@ -63,10 +63,10 @@ Fluss implements a permission inheritance model, where certain operations imply * `READ`, `WRITE`, `CREATE`, `DROP`, `ALTER` each imply `DESCRIBE`. ### Fluss Principal -The FlussPrincipal is a core concept in the Fluss security architecture. It represents the identity of an authenticated entity (such as a user or service) and serves as the central bridge between authentication and authorization. Once a client successfully authenticates via a supported mechanism (e.g., SASL/PLAIN, Kerberos), a FlussPrincipal is created to represent that client's identity. -This principal is then used throughout the system for access control decisions, linking who the user is with what they are allowed to do. +The FlussPrincipal is a core concept in the Fluss security architecture. It represents the identity of an authenticated entity (such as a user or service) and serves as the central bridge between authentication and authorization. Once a client successfully authenticates via a supported mechanism (e.g., SASL/PLAIN, Kerberos), one or more principals represent that client's identity. +These principals are then used throughout the system for access control decisions, linking who the user is with what they are allowed to do. -The principal type indicates the category of the principal (e. g., "User", "Group", "Role"), while the name identifies the specific entity within that category. An authenticated session has one primary principal and may have additional principals such as groups or roles. The default authorizer evaluates ACLs against the primary and additional principals, while superuser checks use only the primary principal. +The principal type indicates the category of the principal (e. g., "User", "Group", "Role"), while the name identifies the specific entity within that category. An authenticated session has a collection of principals, such as a user and its groups or roles. The default authorizer grants access if any principal matches an applicable ACL or a configured superuser. Principal order does not affect authorization. The first principal identifies the client for logging, user metrics, and resource ownership. Example usage: * `new FlussPrincipal("admin", "User")` – A standard user principal. * `new FlussPrincipal("admins", "Group")` – A group-based principal for authorization. @@ -232,3 +232,7 @@ Steps to Implement a Custom Authorization Logic: 2. **Server-Side Plugin Installation**: Build the plugin as a standalone JAR and copy it to the Fluss server’s plugin directory: `/plugins//`. The server will automatically load the plugin at startup. 3. **Configure the Desired Protocol**: Set `org.apache.fluss.server.authorizer.AuthorizationPlugin.identifier` as the value of `authorizer.type` in the Fluss server configuration file. + +### Custom Authentication Plugins + +`ServerAuthenticator.createPrincipals()` returns an `Iterable` containing all authenticated principals. The iterable must be non-null and non-empty, with no null elements. An authenticator that previously implemented `createPrincipal()` must implement `createPrincipals()` instead; single-identity mechanisms can return `Collections.singletonList(principal)`. There is no separate method for group or role principals. diff --git a/website/docs/security/overview.md b/website/docs/security/overview.md index db2bdea234e..bde7f32e09c 100644 --- a/website/docs/security/overview.md +++ b/website/docs/security/overview.md @@ -79,9 +79,9 @@ For example: * User `guest` might only be allowed to consume messages from a specific table. ### Fluss Principal – The Bridge Between Authentication and Authorization -Once a client successfully authenticates, Fluss creates a primary **Fluss Principal**, which represents the authenticated identity. Authentication may also provide additional principals such as groups or roles. +Once a client successfully authenticates, Fluss creates one or more **Fluss Principals**, representing the authenticated user and any groups or roles. -The principal type indicates the category of the principal (e.g., "User", "Group", "Role"), while the name identifies the specific entity within that category. The default authorizer evaluates ACLs against the primary and additional principals, while superuser checks use only the primary principal. +The principal type indicates the category of the principal (e.g., "User", "Group", "Role"), while the name identifies the specific entity within that category. The default authorizer evaluates all authenticated principals equally: access is granted when any principal matches an applicable ACL or a configured superuser. The first principal identifies the client for logging, user metrics, and resource ownership. Example usage: * `new FlussPrincipal("admin", "User")` – A standard user principal. * `new FlussPrincipal("admins", "Group")` – A group-based principal for authorization. From babe783824aec132f136481d34ef2f84e282d57f Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Tue, 10 Nov 2026 16:42:58 +0800 Subject: [PATCH 3/3] [server] Re-run CI for PR 3889