[#9583] improvement(authz): loadTable should indicate if it's for writing

api/src/main/java/org/apache/gravitino/rel/TableCatalog.java

@@ -21,9 +21,11 @@
 package org.apache.gravitino.rel;
 
 import java.util.Map;
+import java.util.Set;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.annotation.Evolving;
+import org.apache.gravitino.authorization.Privilege;
 import org.apache.gravitino.exceptions.NoSuchSchemaException;
 import org.apache.gravitino.exceptions.NoSuchTableException;
 import org.apache.gravitino.exceptions.TableAlreadyExistsException;
@@ -60,6 +62,19 @@ public interface TableCatalog {
    */
   Table loadTable(NameIdentifier ident) throws NoSuchTableException;
 
+  /**
+   * Load table metadata by {@link NameIdentifier} from the catalog with required privileges.
+   *
+   * @param ident A table identifier.
+   * @param requiredPrivilegeNames The required privilege names to access the table.
+   * @return The table metadata.
+   * @throws NoSuchTableException If the table does not exist.
+   */
+  default Table loadTable(NameIdentifier ident, Set<Privilege.Name> requiredPrivilegeNames)
+      throws NoSuchTableException {
+    return loadTable(ident);
+  }
+
   /**
    * Check if a table exists using an {@link NameIdentifier} from the catalog.
    *

clients/client-java/src/main/java/org/apache/gravitino/client/RelationalCatalog.java

@@ -22,17 +22,21 @@
 import static org.apache.gravitino.dto.util.DTOConverters.toDTOs;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.gravitino.Catalog;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
+import org.apache.gravitino.authorization.Privilege;
 import org.apache.gravitino.dto.AuditDTO;
 import org.apache.gravitino.dto.CatalogDTO;
 import org.apache.gravitino.dto.requests.TableCreateRequest;
@@ -61,6 +65,8 @@
  */
 class RelationalCatalog extends BaseSchemaCatalog implements TableCatalog {
 
+  public static final String PRIVILEGES = "privileges";
+
   RelationalCatalog(
       Namespace namespace,
       String name,
@@ -127,6 +133,26 @@ public Table loadTable(NameIdentifier ident) throws NoSuchTableException {
     return RelationalTable.from(fullNamespace, resp.getTable(), restClient);
   }
 
+  @Override
+  public Table loadTable(NameIdentifier ident, Set<Privilege.Name> requiredPrivilegeNames)
+      throws NoSuchTableException {
+    checkTableNameIdentifier(ident);
+
+    Namespace fullNamespace = getTableFullNamespace(ident.namespace());
+    Map<String, String> queryParams = Maps.newHashMap();
+    queryParams.put(PRIVILEGES, Joiner.on(",").join(requiredPrivilegeNames));
+    TableResponse resp =
+        restClient.get(
+            formatTableRequestPath(fullNamespace) + "/" + RESTUtils.encodeString(ident.name()),
+            queryParams,
+            TableResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.tableErrorHandler());
+    resp.validate();
+
+    return RelationalTable.from(fullNamespace, resp.getTable(), restClient);
+  }
+
   /**
    * Create a new table with specified identifier, columns, comment and properties.
    *

clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/authorization/TableAuthorizationIT.java

@@ -23,7 +23,9 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -35,6 +37,7 @@
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.authorization.Owner;
+import org.apache.gravitino.authorization.Privilege;
 import org.apache.gravitino.authorization.Privileges;
 import org.apache.gravitino.authorization.SecurableObject;
 import org.apache.gravitino.authorization.SecurableObjects;
@@ -231,6 +234,15 @@ public void testLoadTable() {
         MetadataObjects.of(ImmutableList.of(CATALOG, SCHEMA), MetadataObject.Type.SCHEMA),
         ImmutableList.of(Privileges.SelectTable.allow()));
     tableCatalogNormalUser.loadTable(NameIdentifier.of(SCHEMA, "table1"));
+
+    assertThrows(
+        String.format("Can not access metadata {%s.%s.%s}.", CATALOG, SCHEMA, "table1"),
+        ForbiddenException.class,
+        () -> {
+          tableCatalogNormalUser.loadTable(
+              NameIdentifier.of(SCHEMA, "table1"), Sets.newHashSet(Privilege.Name.MODIFY_TABLE));
+        });
+
     gravitinoMetalake.grantPrivilegesToRole(
         role,
         MetadataObjects.of(ImmutableList.of(CATALOG, SCHEMA, "table1"), MetadataObject.Type.TABLE),
@@ -241,6 +253,17 @@ public void testLoadTable() {
         () -> {
           tableCatalogNormalUser.loadTable(NameIdentifier.of(SCHEMA, "table1"));
         });
+
+    gravitinoMetalake.grantPrivilegesToRole(
+        role,
+        MetadataObjects.of(ImmutableList.of(CATALOG, SCHEMA, "table1"), MetadataObject.Type.TABLE),
+        ImmutableList.of(Privileges.ModifyTable.allow()));
+    tableCatalogNormalUser.loadTable(
+        NameIdentifier.of(SCHEMA, "table1"), Sets.newHashSet(Privilege.Name.MODIFY_TABLE));
+    gravitinoMetalake.revokePrivilegesFromRole(
+        role,
+        MetadataObjects.of(ImmutableList.of(CATALOG, SCHEMA, "table1"), MetadataObject.Type.TABLE),
+        ImmutableSet.of(Privileges.SelectTable.deny(), Privileges.ModifyTable.allow()));
   }
 
   @Test

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java

@@ -166,7 +166,7 @@ private static CredentialPrivilege getCredentialPrivilege(
         MetadataAuthzHelper.checkAccess(
             identifier,
             Entity.EntityType.TABLE,
-            AuthorizationExpressionConstants.filterModifyTableAuthorizationExpression);
+            AuthorizationExpressionConstants.FILTER_MODIFY_TABLE_AUTHORIZATION_EXPRESSION);
 
     return writable ? CredentialPrivilege.WRITE : CredentialPrivilege.READ;
   }

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergNamespaceOperations.java

@@ -93,7 +93,7 @@ public IcebergNamespaceOperations(
   @Timed(name = "list-namespace." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
   @ResponseMetered(name = "list-namespace", absolute = true)
   @AuthorizationExpression(
-      expression = AuthorizationExpressionConstants.loadCatalogAuthorizationExpression,
+      expression = AuthorizationExpressionConstants.LOAD_CATALOG_AUTHORIZATION_EXPRESSION,
       accessMetadataType = MetadataObject.Type.CATALOG)
   public Response listNamespaces(
       @DefaultValue("") @Encoded() @QueryParam("parent") String parent,
@@ -343,7 +343,7 @@ private ListNamespacesResponse filterListNamespacesResponse(
     NameIdentifier[] idents =
         MetadataAuthzHelper.filterByExpression(
             metalake,
-            AuthorizationExpressionConstants.filterSchemaAuthorizationExpression,
+            AuthorizationExpressionConstants.FILTER_SCHEMA_AUTHORIZATION_EXPRESSION,
             Entity.EntityType.SCHEMA,
             toNameIdentifiers(listNamespacesResponse, metalake, catalogName));
     List<Namespace> filteredNamespaces = new ArrayList<>();

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java

@@ -107,7 +107,7 @@ public IcebergTableOperations(
   @Timed(name = "list-table." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
   @ResponseMetered(name = "list-table", absolute = true)
   @AuthorizationExpression(
-      expression = AuthorizationExpressionConstants.loadSchemaAuthorizationExpression,
+      expression = AuthorizationExpressionConstants.LOAD_SCHEMA_AUTHORIZATION_EXPRESSION,
       accessMetadataType = MetadataObject.Type.SCHEMA)
   public Response listTable(
       @AuthorizationMetadata(type = Entity.EntityType.CATALOG) @PathParam("prefix") String prefix,
@@ -535,7 +535,7 @@ private ListTablesResponse filterListTablesResponse(
     NameIdentifier[] idents =
         MetadataAuthzHelper.filterByExpression(
             metalake,
-            AuthorizationExpressionConstants.filterTableAuthorizationExpression,
+            AuthorizationExpressionConstants.FILTER_TABLE_AUTHORIZATION_EXPRESSION,
             Entity.EntityType.TABLE,
             toNameIdentifiers(listTablesResponse, metalake, catalogName));
     List<TableIdentifier> filteredIdentifiers = new ArrayList<>();

server-common/src/main/java/org/apache/gravitino/server/authorization/annotations/AuthorizationExpression.java

@@ -51,4 +51,18 @@
    * @return error message
    */
   String errorMessage() default "";
+
+  /**
+   * The secondary expression to evaluate for authorization, which represents multiple privileges.
+   *
+   * @return the secondary expression to evaluate for authorization.
+   */
+  String secondaryExpression() default "";
+
+  /**
+   * The condition under which the secondary expression is evaluated.
+   *
+   * @return the condition for evaluating the secondary expression.
+   */
+  String secondaryExpressionCondition() default "";
 }

server-common/src/main/java/org/apache/gravitino/server/authorization/annotations/AuthorizationRequest.java

@@ -32,6 +32,7 @@ enum RequestType {
     COMMON,
     ASSOCIATE_TAG,
     ASSOCIATE_POLICY,
-    RUN_JOB
+    RUN_JOB,
+    LOAD_TABLE
   }
 }

server-common/src/main/java/org/apache/gravitino/server/authorization/expression/AuthorizationExpressionConstants.java

@@ -17,90 +17,90 @@
 package org.apache.gravitino.server.authorization.expression;
 
 public class AuthorizationExpressionConstants {
-  public static final String loadCatalogAuthorizationExpression =
+  public static final String LOAD_CATALOG_AUTHORIZATION_EXPRESSION =
       "ANY_USE_CATALOG || ANY(OWNER, METALAKE, CATALOG)";
 
-  public static final String loadSchemaAuthorizationExpression =
+  public static final String LOAD_SCHEMA_AUTHORIZATION_EXPRESSION =
       """
           ANY(OWNER, METALAKE, CATALOG) ||
           ANY_USE_CATALOG && (SCHEMA::OWNER || ANY_USE_SCHEMA)
            """;
 
-  public static final String loadModelAuthorizationExpression =
+  public static final String LOAD_MODEL_AUTHORIZATION_EXPRESSION =
       """
             ANY(OWNER, METALAKE, CATALOG) ||
              SCHEMA_OWNER_WITH_USE_CATALOG ||
               ANY_USE_CATALOG && ANY_USE_SCHEMA && (MODEL::OWNER || ANY_USE_MODEL)
                   """;
 
-  public static final String loadTableAuthorizationExpression =
+  public static final String LOAD_TABLE_AUTHORIZATION_EXPRESSION =
       """
                   ANY(OWNER, METALAKE, CATALOG) ||
                   SCHEMA_OWNER_WITH_USE_CATALOG ||
                   ANY_USE_CATALOG && ANY_USE_SCHEMA  && (TABLE::OWNER || ANY_SELECT_TABLE || ANY_MODIFY_TABLE)
                   """;
 
-  public static final String alterTableAuthorizationExpression =
+  public static final String MODIFY_TABLE_AUTHORIZATION_EXPRESSION =
       """
                   ANY(OWNER, METALAKE, CATALOG) ||
                   SCHEMA_OWNER_WITH_USE_CATALOG ||
                   ANY_USE_CATALOG && ANY_USE_SCHEMA && (TABLE::OWNER || ANY_MODIFY_TABLE)
                   """;
 
-  public static final String loadTopicsAuthorizationExpression =
+  public static final String LOAD_TOPICS_AUTHORIZATION_EXPRESSION =
       """
           ANY(OWNER, METALAKE, CATALOG) ||
           SCHEMA_OWNER_WITH_USE_CATALOG ||
           ANY_USE_CATALOG && ANY_USE_SCHEMA && (TOPIC::OWNER || ANY_CONSUME_TOPIC || ANY_PRODUCE_TOPIC)
           """;
 
-  public static final String loadFilesetAuthorizationExpression =
+  public static final String LOAD_FILESET_AUTHORIZATION_EXPRESSION =
       """
                  ANY(OWNER, METALAKE, CATALOG) ||
                  SCHEMA_OWNER_WITH_USE_CATALOG ||
                  ANY_USE_CATALOG && ANY_USE_SCHEMA && (FILESET::OWNER || ANY_READ_FILESET || ANY_WRITE_FILESET)
                   """;
 
-  public static final String filterSchemaAuthorizationExpression =
+  public static final String FILTER_SCHEMA_AUTHORIZATION_EXPRESSION =
       "ANY(OWNER, METALAKE, CATALOG, SCHEMA) || ANY_USE_SCHEMA";
 
-  public static final String filterModelAuthorizationExpression =
+  public static final String FILTER_MODEL_AUTHORIZATION_EXPRESSION =
       "ANY(OWNER, METALAKE, CATALOG, SCHEMA, MODEL) || ANY_USE_MODEL";
 
-  public static final String filterTableAuthorizationExpression =
+  public static final String FILTER_TABLE_AUTHORIZATION_EXPRESSION =
       """
                   ANY(OWNER, METALAKE, CATALOG, SCHEMA, TABLE) ||
                   ANY_SELECT_TABLE ||
                   ANY_MODIFY_TABLE
                   """;
 
-  public static final String filterModifyTableAuthorizationExpression =
+  public static final String FILTER_MODIFY_TABLE_AUTHORIZATION_EXPRESSION =
       """
                   ANY(OWNER, METALAKE, CATALOG, SCHEMA, TABLE) ||
                   ANY_MODIFY_TABLE
                   """;
 
-  public static final String filterWriteFilesetAuthorizationExpression =
+  public static final String FILTER_WRITE_FILESET_AUTHORIZATION_EXPRESSION =
       """
                   ANY(OWNER, METALAKE, CATALOG, SCHEMA, FILESET) ||
                   ANY_WRITE_FILESET
                   """;
 
-  public static final String filterTopicsAuthorizationExpression =
+  public static final String FILTER_TOPICS_AUTHORIZATION_EXPRESSION =
       """
               ANY(OWNER, METALAKE, CATALOG, SCHEMA, TOPIC) ||
               ANY_CONSUME_TOPIC ||
               ANY_PRODUCE_TOPIC
        """;
 
-  public static final String filterFilesetAuthorizationExpression =
+  public static final String FILTER_FILESET_AUTHORIZATION_EXPRESSION =
       """
               ANY(OWNER, METALAKE, CATALOG, SCHEMA, FILESET) ||
               ANY_READ_FILESET ||
               ANY_WRITE_FILESET
                   """;
 
-  public static final String loadRoleAuthorizationExpression =
+  public static final String LOAD_ROLE_AUTHORIZATION_EXPRESSION =
       """
           METALAKE::OWNER || METALAKE::MANAGE_GRANTS
           || ROLE::OWNER || ROLE::SELF
@@ -114,13 +114,13 @@ public class AuthorizationExpressionConstants {
           ((CAN_ACCESS_METADATA) && (TAG::OWNER || ANY_APPLY_TAG))
           """;
 
-  public static final String loadTagAuthorizationExpression =
+  public static final String LOAD_TAG_AUTHORIZATION_EXPRESSION =
       "METALAKE::OWNER || TAG::OWNER || ANY_APPLY_TAG";
 
-  public static final String applyTagAuthorizationExpression =
+  public static final String APPLY_TAG_AUTHORIZATION_EXPRESSION =
       "METALAKE::OWNER || TAG::OWNER || ANY_APPLY_TAG";
 
-  public static final String loadPolicyAuthorizationExpression =
+  public static final String LOAD_POLICY_AUTHORIZATION_EXPRESSION =
       """
           METALAKE::OWNER || POLICY::OWNER || ANY_APPLY_POLICY
           """;
@@ -131,10 +131,13 @@ public class AuthorizationExpressionConstants {
    * authorization checks {@link
    * org.apache.gravitino.authorization.GravitinoAuthorizer#isMetalakeUser(String)}.
    */
-  public static final String loadMetalakeAuthorizationExpression = "METALAKE_USER";
+  public static final String LOAD_METALAKE_AUTHORIZATION_EXPRESSION = "METALAKE_USER";
 
-  public static final String loadJobAuthorizationExpression = "METALAKE::OWNER || JOB::OWNER";
+  public static final String LOAD_JOB_AUTHORIZATION_EXPRESSION = "METALAKE::OWNER || JOB::OWNER";
 
-  public static final String loadJobTemplateAuthorizationExpression =
+  public static final String LOAD_JOB_TEMPLATE_AUTHORIZATION_EXPRESSION =
       "METALAKE::OWNER || JOB_TEMPLATE::OWNER || ANY_USE_JOB_TEMPLATE";
+
+  public static final String REQUEST_REQUIRED_PRIVILEGES_CONTAINS_MODIFY_TABLE =
+      "REQUEST::REQUIRED_PRIVILEGES_CONTAINS_MODIFY_TABLE";
 }