[#12112] fix(authz): Align group load authorization

server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java

@@ -424,10 +424,27 @@ public boolean isSelf(
       NameIdentifier nameIdentifier,
       AuthorizationRequestContext requestContext) {
     String metalake = nameIdentifier.namespace().level(0);
-    String currentUserName = PrincipalUtils.getCurrentUserName();
     if (Entity.EntityType.USER == type) {
+      String currentUserName = PrincipalUtils.getCurrentUserName();
       return Objects.equals(nameIdentifier.name(), currentUserName);
+    } else if (Entity.EntityType.GROUP == type) {
+      Principal currentPrincipal = PrincipalUtils.getCurrentPrincipal();
+      if (!(currentPrincipal instanceof UserPrincipal)) {
+        return false;
+      }
+
+      List<UserGroup> groups = ((UserPrincipal) currentPrincipal).getGroups();
+      if (groups.isEmpty()) {
+        return false;
+      }
+
+      boolean principalHasGroup =
+          groups.stream()
+              .map(UserGroup::getGroupName)
+              .anyMatch(groupName -> Objects.equals(groupName, nameIdentifier.name()));
+      return principalHasGroup;
     } else if (Entity.EntityType.ROLE == type) {
+      String currentUserName = PrincipalUtils.getCurrentUserName();
       try {
         Optional<Long> roleId =
             MetadataIdConverter.getID(

server-common/src/test/java/org/apache/gravitino/server/authorization/jcasbin/TestJcasbinAuthorizer.java

@@ -886,6 +886,46 @@ public void testIsSelfRoleViaGroup() throws Exception {
     restoreDefaultPrincipal();
   }
 
+  @Test
+  public void testIsSelfGroupViaPrincipalGroup() throws Exception {
+    NameIdentifier groupIdent = NameIdentifierUtil.ofGroup(METALAKE, GROUP_NAME);
+
+    setCurrentPrincipalWithGroup(GROUP_NAME);
+    Mockito.clearInvocations(groupMetaMapper);
+    assertTrue(
+        jcasbinAuthorizer.isSelf(
+            Entity.EntityType.GROUP, groupIdent, new AuthorizationRequestContext()));
+    Mockito.verify(groupMetaMapper, Mockito.never()).getGroupUpdatedAt(anyString(), anyString());
+
+    setCurrentPrincipalWithGroup("otherGroup");
+    assertFalse(
+        jcasbinAuthorizer.isSelf(
+            Entity.EntityType.GROUP, groupIdent, new AuthorizationRequestContext()));
+    Mockito.verify(groupMetaMapper, Mockito.never()).getGroupUpdatedAt(anyString(), anyString());
+
+    restoreDefaultPrincipal();
+  }
+
+  @Test
+  public void testIsSelfGroupDoesNotRequireCurrentUserName() throws Exception {
+    NameIdentifier groupIdent = NameIdentifierUtil.ofGroup(METALAKE, GROUP_NAME);
+
+    UserPrincipal groupPrincipal = mock(UserPrincipal.class);
+    when(groupPrincipal.getGroups())
+        .thenReturn(ImmutableList.of(new UserGroup(Optional.empty(), GROUP_NAME)));
+    when(groupPrincipal.getName())
+        .thenThrow(new AssertionError("GROUP self check should only use principal groups"));
+    principalUtilsMockedStatic.when(PrincipalUtils::getCurrentPrincipal).thenReturn(groupPrincipal);
+
+    try {
+      assertTrue(
+          jcasbinAuthorizer.isSelf(
+              Entity.EntityType.GROUP, groupIdent, new AuthorizationRequestContext()));
+    } finally {
+      restoreDefaultPrincipal();
+    }
+  }
+
   @Test
   public void testIsSelfRoleReusesCacheAcrossCalls() throws Exception {
     // Acceptance criterion for #11088: repeated isSelf(ROLE) calls in the same logical request

server/src/main/java/org/apache/gravitino/server/web/rest/GroupOperations.java

@@ -49,10 +49,12 @@
 import org.apache.gravitino.dto.util.DTOConverters;
 import org.apache.gravitino.metalake.MetalakeManager;
 import org.apache.gravitino.metrics.MetricNames;
+import org.apache.gravitino.server.authorization.MetadataAuthzHelper;
 import org.apache.gravitino.server.authorization.NameBindings;
 import org.apache.gravitino.server.authorization.annotations.AuthorizationExpression;
 import org.apache.gravitino.server.authorization.annotations.AuthorizationMetadata;
 import org.apache.gravitino.server.web.Utils;
+import org.apache.gravitino.utils.NameIdentifierUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -62,6 +64,9 @@ public class GroupOperations {
 
   private static final Logger LOG = LoggerFactory.getLogger(GroupOperations.class);
 
+  private static final String LOAD_GROUP_PRIVILEGE =
+      "METALAKE::OWNER || METALAKE::MANAGE_GROUPS || GROUP::SELF";
+
   private final AccessControlDispatcher accessControlManager;
   private final OwnerDispatcher ownerDispatcher;
 
@@ -80,8 +85,11 @@ public GroupOperations() {
   @Produces("application/vnd.gravitino.v1+json")
   @Timed(name = "get-group." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
   @ResponseMetered(name = "get-group", absolute = true)
+  @AuthorizationExpression(expression = LOAD_GROUP_PRIVILEGE)
   public Response getGroup(
-      @PathParam("metalake") String metalake, @PathParam("group") String group) {
+      @PathParam("metalake") @AuthorizationMetadata(type = Entity.EntityType.METALAKE)
+          String metalake,
+      @PathParam("group") @AuthorizationMetadata(type = Entity.EntityType.GROUP) String group) {
     try {
       return Utils.doAs(
           httpRequest,
@@ -179,11 +187,26 @@ public Response listGroups(
           () -> {
             MetalakeManager.checkMetalakeInUse(metalake);
             if (verbose) {
-              return Utils.ok(
-                  new GroupListResponse(
-                      DTOConverters.toDTOs(accessControlManager.listGroups(metalake))));
+              Group[] groups = accessControlManager.listGroups(metalake);
+              groups =
+                  MetadataAuthzHelper.filterByExpression(
+                      metalake,
+                      LOAD_GROUP_PRIVILEGE,
+                      Entity.EntityType.GROUP,
+                      groups,
+                      groupEntity -> NameIdentifierUtil.ofGroup(metalake, groupEntity.name()));
+
+              return Utils.ok(new GroupListResponse(DTOConverters.toDTOs(groups)));
             } else {
-              return Utils.ok(new NameListResponse(accessControlManager.listGroupNames(metalake)));
+              String[] groups = accessControlManager.listGroupNames(metalake);
+              groups =
+                  MetadataAuthzHelper.filterByExpression(
+                      metalake,
+                      LOAD_GROUP_PRIVILEGE,
+                      Entity.EntityType.GROUP,
+                      groups,
+                      groupName -> NameIdentifierUtil.ofGroup(metalake, groupName));
+              return Utils.ok(new NameListResponse(groups));
             }
           });
 

server/src/test/java/org/apache/gravitino/server/web/rest/TestGroupOperations.java

@@ -38,6 +38,7 @@
 import javax.ws.rs.core.Response.Status;
 import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.gravitino.Config;
+import org.apache.gravitino.Entity.EntityType;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.GravitinoEnv;
 import org.apache.gravitino.authorization.AccessControlManager;
@@ -62,12 +63,14 @@
 import org.apache.gravitino.meta.BaseMetalake;
 import org.apache.gravitino.meta.GroupEntity;
 import org.apache.gravitino.rest.RESTUtils;
+import org.apache.gravitino.server.authorization.MetadataAuthzHelper;
 import org.glassfish.hk2.utilities.binding.AbstractBinder;
 import org.glassfish.jersey.server.ResourceConfig;
 import org.glassfish.jersey.test.TestProperties;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 
 public class TestGroupOperations extends BaseOperationsTest {
@@ -380,6 +383,43 @@ public void testListGroupNames() throws IOException {
     Assertions.assertEquals(RuntimeException.class.getSimpleName(), errorResponse2.getType());
   }
 
+  @Test
+  public void testListGroupNamesFiltersByAuthorizationExpression() throws Exception {
+    Mockito.reset(manager, entityStore);
+    Mockito.doReturn(new String[] {"group", "filtered"}).when(manager).listGroupNames(any());
+
+    // Mock metalake with in-use property
+    BaseMetalake metalake = mock(BaseMetalake.class);
+    PropertiesMetadata propertiesMetadata = mock(PropertiesMetadata.class);
+    when(propertiesMetadata.getOrDefault(any(), any())).thenReturn(true);
+    when(metalake.propertiesMetadata()).thenReturn(propertiesMetadata);
+    when(entityStore.get(any(), any(), any())).thenReturn(metalake);
+
+    try (MockedStatic<MetadataAuthzHelper> metadataAuthzHelper =
+        Mockito.mockStatic(MetadataAuthzHelper.class)) {
+      metadataAuthzHelper
+          .when(
+              () ->
+                  MetadataAuthzHelper.filterByExpression(
+                      Mockito.eq("metalake1"),
+                      Mockito.eq("METALAKE::OWNER || METALAKE::MANAGE_GROUPS || GROUP::SELF"),
+                      Mockito.eq(EntityType.GROUP),
+                      Mockito.any(String[].class),
+                      Mockito.any()))
+          .thenReturn(new String[] {"group"});
+
+      GroupOperations groupOperations = new GroupOperations();
+      HttpServletRequest request = mock(HttpServletRequest.class);
+      FieldUtils.writeField(groupOperations, "httpRequest", request, true);
+
+      Response resp = groupOperations.listGroups("metalake1", false);
+
+      Assertions.assertEquals(Response.Status.OK.getStatusCode(), resp.getStatus());
+      NameListResponse listResponse = (NameListResponse) resp.getEntity();
+      Assertions.assertArrayEquals(new String[] {"group"}, listResponse.getNames());
+    }
+  }
+
   @Test
   public void testListGroups() throws IOException {
     Group group = buildGroup("group");
@@ -445,6 +485,46 @@ public void testListGroups() throws IOException {
     Assertions.assertEquals(RuntimeException.class.getSimpleName(), errorResponse2.getType());
   }
 
+  @Test
+  public void testListGroupsFiltersByAuthorizationExpression() throws Exception {
+    Mockito.reset(manager, entityStore);
+    Group group = buildGroup("group");
+    Group filteredGroup = buildGroup("filtered");
+    Mockito.doReturn(new Group[] {group, filteredGroup}).when(manager).listGroups(any());
+
+    // Mock metalake with in-use property
+    BaseMetalake metalake = mock(BaseMetalake.class);
+    PropertiesMetadata propertiesMetadata = mock(PropertiesMetadata.class);
+    when(propertiesMetadata.getOrDefault(any(), any())).thenReturn(true);
+    when(metalake.propertiesMetadata()).thenReturn(propertiesMetadata);
+    when(entityStore.get(any(), any(), any())).thenReturn(metalake);
+
+    try (MockedStatic<MetadataAuthzHelper> metadataAuthzHelper =
+        Mockito.mockStatic(MetadataAuthzHelper.class)) {
+      metadataAuthzHelper
+          .when(
+              () ->
+                  MetadataAuthzHelper.filterByExpression(
+                      Mockito.eq("metalake1"),
+                      Mockito.eq("METALAKE::OWNER || METALAKE::MANAGE_GROUPS || GROUP::SELF"),
+                      Mockito.eq(EntityType.GROUP),
+                      Mockito.any(Group[].class),
+                      Mockito.any()))
+          .thenReturn(new Group[] {group});
+
+      GroupOperations groupOperations = new GroupOperations();
+      HttpServletRequest request = mock(HttpServletRequest.class);
+      FieldUtils.writeField(groupOperations, "httpRequest", request, true);
+
+      Response resp = groupOperations.listGroups("metalake1", true);
+
+      Assertions.assertEquals(Response.Status.OK.getStatusCode(), resp.getStatus());
+      GroupListResponse listResponse = (GroupListResponse) resp.getEntity();
+      Assertions.assertEquals(1, listResponse.getGroups().length);
+      Assertions.assertEquals(group.name(), listResponse.getGroups()[0].name());
+    }
+  }
+
   @Test
   public void testRemoveGroup() throws IOException {
     // Mock metalake with in-use property

server/src/test/java/org/apache/gravitino/server/web/rest/authorization/TestGroupAuthorizationExpression.java

@@ -0,0 +1,66 @@
+/*
+ * 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.gravitino.server.web.rest.authorization;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.common.collect.ImmutableSet;
+import java.lang.reflect.Method;
+import java.lang.reflect.Parameter;
+import ognl.OgnlException;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.server.authorization.annotations.AuthorizationExpression;
+import org.apache.gravitino.server.authorization.annotations.AuthorizationMetadata;
+import org.apache.gravitino.server.web.rest.GroupOperations;
+import org.junit.jupiter.api.Test;
+
+public class TestGroupAuthorizationExpression {
+
+  @Test
+  public void testGetGroupAuthorizationExpression() throws NoSuchMethodException, OgnlException {
+    Method method = GroupOperations.class.getMethod("getGroup", String.class, String.class);
+    AuthorizationExpression authorizationExpressionAnnotation =
+        method.getAnnotation(AuthorizationExpression.class);
+    assertNotNull(authorizationExpressionAnnotation);
+
+    MockAuthorizationExpressionEvaluator mockEvaluator =
+        new MockAuthorizationExpressionEvaluator(authorizationExpressionAnnotation.expression());
+    assertFalse(mockEvaluator.getResult(ImmutableSet.of()));
+    assertFalse(mockEvaluator.getResult(ImmutableSet.of("METALAKE::MANAGE_USERS")));
+    assertTrue(mockEvaluator.getResult(ImmutableSet.of("METALAKE::OWNER")));
+    assertTrue(mockEvaluator.getResult(ImmutableSet.of("METALAKE::MANAGE_GROUPS")));
+    assertTrue(mockEvaluator.getResult(ImmutableSet.of("GROUP::SELF")));
+  }
+
+  @Test
+  public void testGetGroupAuthorizationMetadata() throws NoSuchMethodException {
+    Method method = GroupOperations.class.getMethod("getGroup", String.class, String.class);
+    Parameter[] parameters = method.getParameters();
+
+    AuthorizationMetadata metalakeMetadata =
+        parameters[0].getAnnotation(AuthorizationMetadata.class);
+    assertNotNull(metalakeMetadata);
+    assertEquals(Entity.EntityType.METALAKE, metalakeMetadata.type());
+
+    AuthorizationMetadata groupMetadata = parameters[1].getAnnotation(AuthorizationMetadata.class);
+    assertNotNull(groupMetadata);
+    assertEquals(Entity.EntityType.GROUP, groupMetadata.type());
+  }
+}