From 5892cdda73ad88c32cac198ecd081ec62ab0c33a Mon Sep 17 00:00:00 2001
From: vaibhav kumar <vaibhakumar@expediagroup.com>
Date: Mon, 11 May 2026 15:19:58 +0530
Subject: [PATCH] [server] Add authorization to listOffsets for TABLE/DESCRIBE
 operations

Implements authorization checks for the listOffsets method as part of issue #3247.
This completes authorization for all three table metadata read operations:
- getTableSchema (already had authorization)
- listPartitionInfos (already had authorization)
- listOffsets (added in this commit)

Changes:
- Added authorizeTable(DESCRIBE, tableId) check in TabletService.listOffsets()
- Removed TODO comment indicating this work was needed
- Added comprehensive authorization tests in FlussAuthorizationITCase
- Tests verify both authorization denial and success scenarios

This ensures that clients must have DESCRIBE permission on a table before
they can query offset information, preventing unauthorized access to table
metadata.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
---
 .../acl/FlussAuthorizationITCase.java         | 20 +++++++++++++++++++
 .../fluss/server/tablet/TabletService.java    |  3 ++-
 2 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java
index 0ac000359d..5fc9cb1e76 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java
@@ -23,6 +23,7 @@
 import org.apache.fluss.client.admin.Admin;
 import org.apache.fluss.client.admin.FlussAdmin;
 import org.apache.fluss.client.admin.KvSnapshotLease;
+import org.apache.fluss.client.admin.OffsetSpec;
 import org.apache.fluss.client.table.Table;
 import org.apache.fluss.client.table.scanner.batch.BatchScanner;
 import org.apache.fluss.client.table.writer.AppendWriter;
@@ -416,6 +417,7 @@ void testDescribeTableOperation() throws Exception {
         // 4. getLatestKvSnapshots
         // 5. listPartitionInfos
         // 6. getLatestLakeSnapshot
+        // 7. listOffsets
 
         // first check call these methods without authorization.
         assertThat(guestAdmin.listTables(DATA1_TABLE_PATH_PK.getDatabaseName()).get())
@@ -426,6 +428,15 @@ void testDescribeTableOperation() throws Exception {
         assertNoTableDescribeAuth(() -> guestAdmin.listPartitionInfos(DATA1_TABLE_PATH_PK).get());
         assertNoTableDescribeAuth(
                 () -> guestAdmin.getLatestLakeSnapshot(DATA1_TABLE_PATH_PK).get());
+        assertNoTableDescribeAuth(
+                () ->
+                        guestAdmin
+                                .listOffsets(
+                                        DATA1_TABLE_PATH_PK,
+                                        Arrays.asList(0),
+                                        new OffsetSpec.LatestSpec())
+                                .all()
+                                .get());
 
         // add acl to allow guest describe table resource
         List<AclBinding> aclBindings =
@@ -459,6 +470,15 @@ void testDescribeTableOperation() throws Exception {
                 .rootCause()
                 .isInstanceOf(LakeTableSnapshotNotExistException.class)
                 .hasMessageContaining("Lake table snapshot doesn't exist for table");
+        assertThat(
+                        guestAdmin
+                                .listOffsets(
+                                        DATA1_TABLE_PATH_PK,
+                                        Arrays.asList(0),
+                                        new OffsetSpec.LatestSpec())
+                                .all()
+                                .get())
+                .isNotEmpty();
     }
 
     @ParameterizedTest
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
index 88731daaba..c18b1c5306 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
@@ -109,6 +109,7 @@
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 
+import static org.apache.fluss.security.acl.OperationType.DESCRIBE;
 import static org.apache.fluss.security.acl.OperationType.READ;
 import static org.apache.fluss.security.acl.OperationType.WRITE;
 import static org.apache.fluss.server.coordinator.CoordinatorContext.INITIAL_COORDINATOR_EPOCH;
@@ -395,7 +396,7 @@ public CompletableFuture<StopReplicaResponse> stopReplica(
 
     @Override
     public CompletableFuture<ListOffsetsResponse> listOffsets(ListOffsetsRequest request) {
-        // TODO: authorize DESCRIBE permission
+        authorizeTable(DESCRIBE, request.getTableId());
         CompletableFuture<ListOffsetsResponse> response = new CompletableFuture<>();
         Set<TableBucket> tableBuckets = getListOffsetsData(request);
         replicaManager.listOffsets(