[#2869] feat(core): Add the core logic of authorization

api/src/main/java/com/datastrato/gravitino/authorization/Role.java

@@ -17,6 +17,8 @@
 @Evolving
 public interface Role extends Auditable {
 
+  /** The reserved role names' prefix */
+  String SYSTEM_RESERVED_ROLE_NAME_PREFIX = "system_role";
   /**
    * The name of the role.
    *

api/src/main/java/com/datastrato/gravitino/authorization/SecurableObjects.java

@@ -104,8 +104,8 @@ public static SecurableObject ofFileset(
   /**
    * All metalakes is a special securable object .You can give the securable object the privileges
    * `CREATE METALAKE`, etc. It means that you can create any which doesn't exist. This securable
-   * object is only used for metalake admin. You can't grant any privilege to this securable object.
-   * You can't bind this securable object to any role, too.
+   * object is reserved for system. You can't grant any privilege to this securable object. You
+   * can't bind this securable object to any role, too.
    *
    * @param privileges The privileges of the all metalakes
    * @return The created {@link SecurableObject}

api/src/main/java/com/datastrato/gravitino/exceptions/ForbiddenException.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** Exception thrown when a user doesn't have the privilege to perform an action. */
+public class ForbiddenException extends GravitinoRuntimeException {
+
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public ForbiddenException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+
+  /**
+   * Constructs a new exception with the specified detail message and cause.
+   *
+   * @param cause the cause.
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public ForbiddenException(Throwable cause, @FormatString String message, Object... args) {
+    super(cause, message, args);
+  }
+}

api/src/main/java/com/datastrato/gravitino/exceptions/PrivilegesAlreadyGrantedException.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** An exception thrown when privileges already exist. */
+public class PrivilegesAlreadyGrantedException extends AlreadyExistsException {
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public PrivilegesAlreadyGrantedException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+
+  /**
+   * Constructs a new exception with the specified detail message and cause.
+   *
+   * @param cause the cause.
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public PrivilegesAlreadyGrantedException(
+      Throwable cause, @FormatString String message, Object... args) {
+    super(cause, message, args);
+  }
+}

clients/client-java/src/main/java/com/datastrato/gravitino/client/ErrorHandlers.java

@@ -25,6 +25,7 @@
 import com.datastrato.gravitino.exceptions.NonEmptySchemaException;
 import com.datastrato.gravitino.exceptions.NotFoundException;
 import com.datastrato.gravitino.exceptions.PartitionAlreadyExistsException;
+import com.datastrato.gravitino.exceptions.PrivilegesAlreadyGrantedException;
 import com.datastrato.gravitino.exceptions.RESTException;
 import com.datastrato.gravitino.exceptions.RoleAlreadyExistsException;
 import com.datastrato.gravitino.exceptions.SchemaAlreadyExistsException;
@@ -143,6 +144,15 @@ public static Consumer<ErrorResponse> groupErrorHandler() {
     return GroupErrorHandler.INSTANCE;
   }
 
+  /**
+   * Creates an error handler specific to MetalakeAdmin operations.
+   *
+   * @return A Consumer representing the MetalakeAdmin error handler.
+   */
+  public static Consumer<ErrorResponse> metalakeAdminErrorHandler() {
+    return MetalakeAdminErrorHandler.INSTANCE;
+  }
+
   /**
    * Creates an error handler specific to Role operations.
    *
@@ -567,6 +577,41 @@ public void accept(ErrorResponse errorResponse) {
     }
   }
 
+  /** Error handler specific to MetalakeAdmin operations. */
+  @SuppressWarnings("FormatStringAnnotation")
+  private static class MetalakeAdminErrorHandler extends RestErrorHandler {
+
+    private static final MetalakeAdminErrorHandler INSTANCE = new MetalakeAdminErrorHandler();
+
+    @Override
+    public void accept(ErrorResponse errorResponse) {
+      String errorMessage = formatErrorMessage(errorResponse);
+
+      switch (errorResponse.getCode()) {
+        case ErrorConstants.ILLEGAL_ARGUMENTS_CODE:
+          throw new IllegalArgumentException(errorMessage);
+
+        case ErrorConstants.NOT_FOUND_CODE:
+          if (errorResponse.getType().equals(NoSuchMetalakeException.class.getSimpleName())) {
+            throw new NoSuchMetalakeException(errorMessage);
+          } else if (errorResponse.getType().equals(NoSuchUserException.class.getSimpleName())) {
+            throw new NoSuchUserException(errorMessage);
+          } else {
+            throw new NotFoundException(errorMessage);
+          }
+
+        case ErrorConstants.ALREADY_EXISTS_CODE:
+          throw new PrivilegesAlreadyGrantedException(errorMessage);
+
+        case ErrorConstants.INTERNAL_ERROR_CODE:
+          throw new RuntimeException(errorMessage);
+
+        default:
+          super.accept(errorResponse);
+      }
+    }
+  }
+
   /** Error handler specific to Role operations. */
   @SuppressWarnings("FormatStringAnnotation")
   private static class RoleErrorHandler extends RestErrorHandler {

clients/client-java/src/main/java/com/datastrato/gravitino/client/GravitinoAdminClient.java

@@ -34,6 +34,7 @@
 import com.datastrato.gravitino.exceptions.NoSuchMetalakeException;
 import com.datastrato.gravitino.exceptions.NoSuchRoleException;
 import com.datastrato.gravitino.exceptions.NoSuchUserException;
+import com.datastrato.gravitino.exceptions.PrivilegesAlreadyGrantedException;
 import com.datastrato.gravitino.exceptions.RoleAlreadyExistsException;
 import com.datastrato.gravitino.exceptions.UserAlreadyExistsException;
 import com.google.common.base.Preconditions;
@@ -325,10 +326,11 @@ public Group getGroup(String metalake, String group)
    *
    * @param user The name of the User.
    * @return The added User instance.
-   * @throws UserAlreadyExistsException If a metalake admin with the same name already exists.
+   * @throws PrivilegesAlreadyGrantedException If the metalake admin with the same name already
+   *     added.
    * @throws RuntimeException If adding the User encounters storage issues.
    */
-  public User addMetalakeAdmin(String user) throws UserAlreadyExistsException {
+  public User addMetalakeAdmin(String user) throws PrivilegesAlreadyGrantedException {
     UserAddRequest req = new UserAddRequest(user);
     req.validate();
 
@@ -338,7 +340,7 @@ public User addMetalakeAdmin(String user) throws UserAlreadyExistsException {
             req,
             UserResponse.class,
             Collections.emptyMap(),
-            ErrorHandlers.userErrorHandler());
+            ErrorHandlers.metalakeAdminErrorHandler());
     resp.validate();
 
     return resp.getUser();
@@ -358,7 +360,7 @@ public boolean removeMetalakeAdmin(String user) {
             String.format(API_ADMIN_PATH, user),
             RemoveResponse.class,
             Collections.emptyMap(),
-            ErrorHandlers.userErrorHandler());
+            ErrorHandlers.metalakeAdminErrorHandler());
     resp.validate();
 
     return resp.removed();

clients/client-java/src/test/java/com/datastrato/gravitino/client/TestMetalakeAdmin.java

@@ -15,7 +15,7 @@
 import com.datastrato.gravitino.dto.responses.ErrorResponse;
 import com.datastrato.gravitino.dto.responses.RemoveResponse;
 import com.datastrato.gravitino.dto.responses.UserResponse;
-import com.datastrato.gravitino.exceptions.UserAlreadyExistsException;
+import com.datastrato.gravitino.exceptions.PrivilegesAlreadyGrantedException;
 import java.time.Instant;
 import org.apache.hc.core5.http.Method;
 import org.junit.jupiter.api.Assertions;
@@ -45,14 +45,14 @@ public void testAddMetalakeAdmin() throws Exception {
     Assertions.assertNotNull(addedUser);
     assertUser(addedUser, mockUser);
 
-    // test UserAlreadyExistsException
+    // test PrivilegeGrantAlreadyExistsException
     ErrorResponse errResp1 =
         ErrorResponse.alreadyExists(
-            UserAlreadyExistsException.class.getSimpleName(), "user already exists");
+            PrivilegesAlreadyGrantedException.class.getSimpleName(), "user already exists");
     buildMockResource(Method.POST, userPath, request, errResp1, SC_CONFLICT);
     Exception ex =
         Assertions.assertThrows(
-            UserAlreadyExistsException.class, () -> client.addMetalakeAdmin(username));
+            PrivilegesAlreadyGrantedException.class, () -> client.addMetalakeAdmin(username));
     Assertions.assertEquals("user already exists", ex.getMessage());
 
     // test RuntimeException

core/src/main/java/com/datastrato/gravitino/Entity.java

@@ -23,9 +23,6 @@ public interface Entity extends Serializable {
   /** The system reserved catalog name. */
   String SYSTEM_CATALOG_RESERVED_NAME = "system";
 
-  /** The authorization catalog name in the system metalake. */
-  String AUTHORIZATION_CATALOG_NAME = "authorization";
-
   /** The user schema name in the system catalog. */
   String USER_SCHEMA_NAME = "user";
 
@@ -35,15 +32,12 @@ public interface Entity extends Serializable {
   /** The role schema name in the system catalog. */
   String ROLE_SCHEMA_NAME = "role";
 
-  /** The admin schema name in the authorization catalog of the system metalake. */
-  String ADMIN_SCHEMA_NAME = "admin";
-
   /** The tag schema name in the system catalog. */
   String TAG_SCHEMA_NAME = "tag";
 
   /**
    * All metalakes are a virtual entity. It represents all the metalakes. We don't store it. We use
-   * a specific type to represent its entity type.
+   * a specific type to represent its entity id.
    */
   String ALL_METALAKES_ENTITY_TYPE = "ROOT";
 

core/src/main/java/com/datastrato/gravitino/authorization/AccessControlManager.java

@@ -11,13 +11,16 @@
 import com.datastrato.gravitino.exceptions.NoSuchMetalakeException;
 import com.datastrato.gravitino.exceptions.NoSuchRoleException;
 import com.datastrato.gravitino.exceptions.NoSuchUserException;
+import com.datastrato.gravitino.exceptions.PrivilegesAlreadyGrantedException;
 import com.datastrato.gravitino.exceptions.RoleAlreadyExistsException;
 import com.datastrato.gravitino.exceptions.UserAlreadyExistsException;
+import com.datastrato.gravitino.meta.RoleEntity;
 import com.datastrato.gravitino.storage.IdGenerator;
 import com.datastrato.gravitino.utils.Executable;
 import com.google.common.annotations.VisibleForTesting;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * AccessControlManager is used for manage users, roles, admin, grant information, this class is an
@@ -37,8 +40,8 @@ public class AccessControlManager {
   private final Object nonAdminOperationLock = new Object();
 
   public AccessControlManager(EntityStore store, IdGenerator idGenerator, Config config) {
-    this.adminManager = new AdminManager(store, idGenerator, config);
     this.roleManager = new RoleManager(store, idGenerator, config);
+    this.adminManager = new AdminManager(store, idGenerator, config);
     this.userGroupManager = new UserGroupManager(store, idGenerator, roleManager);
     this.permissionManager = new PermissionManager(store, roleManager);
   }
@@ -204,15 +207,16 @@ public User revokeRolesFromUser(String metalake, List<String> roles, String user
    *
    * @param user The name of the User.
    * @return The added User instance.
-   * @throws UserAlreadyExistsException If a metalake admin with the same name already exists.
+   * @throws PrivilegesAlreadyGrantedException If the metalake admin with the same name already
+   *     added.
    * @throws RuntimeException If adding the User encounters storage issues.
    */
-  public User addMetalakeAdmin(String user) throws UserAlreadyExistsException {
+  public User addMetalakeAdmin(String user) throws PrivilegesAlreadyGrantedException {
     return doWithAdminLock(() -> adminManager.addMetalakeAdmin(user));
   }
 
   /**
-   * Removes a metalake admin.
+   * Removes a metalake admin. Only service admins can manage metalake admins.
    *
    * @param user The name of the User.
    * @return True if the User was successfully removed, false only when there's no such metalake
@@ -233,16 +237,6 @@ public boolean isServiceAdmin(String user) {
     return adminManager.isServiceAdmin(user);
   }
 
-  /**
-   * Judges whether the user is the metalake admin.
-   *
-   * @param user the name of the user
-   * @return True if the user is metalake admin, otherwise false.
-   */
-  public boolean isMetalakeAdmin(String user) {
-    return doWithAdminLock(() -> adminManager.isMetalakeAdmin(user));
-  }
-
   /**
    * Creates a new Role.
    *
@@ -294,6 +288,17 @@ public boolean deleteRole(String metalake, String role) throws NoSuchMetalakeExc
     return doWithNonAdminLock(() -> roleManager.deleteRole(metalake, role));
   }
 
+  public List<RoleEntity> listRolesByUser(String metalake, String currentUser) {
+    return doWithNonAdminLock(
+        () -> {
+          User user = getUser(metalake, currentUser);
+          // TODO: get roles from the group
+          return user.roles().stream()
+              .map(role -> roleManager.getRole(metalake, role))
+              .collect(Collectors.toList());
+        });
+  }
+
   @VisibleForTesting
   RoleManager getRoleManager() {
     return roleManager;