Allow CatalogAdmin to list Principal Roles

integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisManagementServiceIntegrationTest.java

@@ -1957,6 +1957,82 @@ public void testCatalogAdminGrantAndRevokeCatalogRolesFromWrongCatalog() {
     }
   }
 
+  @Test
+  public void testCatalogRoleManagerAutoGrantAndRevoke() {
+    // Create a catalog
+    String catalogName = client.newEntityName("catalog_role_mgr_test");
+    Catalog catalog =
+        PolarisCatalog.builder()
+            .setType(Catalog.TypeEnum.INTERNAL)
+            .setName(catalogName)
+            .setStorageConfigInfo(new AwsStorageConfigInfo(StorageConfigInfo.StorageTypeEnum.S3))
+            .setProperties(new CatalogProperties("s3://bucket1/"))
+            .build();
+    managementApi.createCatalog(catalog);
+
+    // Create a principal and principal role
+    PrincipalWithCredentials principal =
+        managementApi.createPrincipal(client.newEntityName("test_principal"));
+    String principalRoleName = client.newEntityName("test_role");
+    managementApi.createPrincipalRole(principalRoleName);
+
+    // Assign principal to principal role
+    managementApi.assignPrincipalRole(principal.getPrincipal().getName(), principalRoleName);
+
+    // Verify principal does not have catalog_role_manager yet by checking activated roles
+    Principal principalBefore = managementApi.getPrincipal(principal.getPrincipal().getName());
+    assertThat(principalBefore).isNotNull();
+
+    // Grant catalog_admin to the principal role
+    CatalogRole catalogAdminRole =
+        managementApi.getCatalogRole(
+            catalogName, PolarisEntityConstants.getNameOfCatalogAdminRole());
+    managementApi.grantCatalogRoleToPrincipalRole(principalRoleName, catalogName, catalogAdminRole);
+
+    // Verify principal now has catalog_role_manager (auto-granted)
+    try (Response response =
+        managementApi
+            .request(
+                "v1/principals/{p}/principal-roles",
+                Map.of("p", principal.getPrincipal().getName()))
+            .get()) {
+      assertThat(response.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+      PrincipalRoles roles = response.readEntity(PrincipalRoles.class);
+      assertThat(roles.getRoles())
+          .extracting(PrincipalRole::getName)
+          .contains(PolarisEntityConstants.getNameOfCatalogRoleManagerPrincipalRole());
+    }
+
+    // Verify catalog_admin can list principal roles
+    String catalogAdminToken = client.obtainToken(principal);
+    List<PrincipalRole> listedRoles = client.managementApi(catalogAdminToken).listPrincipalRoles();
+    assertThat(listedRoles).isNotEmpty();
+
+    // Revoke catalog_admin from the principal role
+    try (Response response =
+        managementApi
+            .request(
+                "v1/principal-roles/{pr}/catalog-roles/{cat}/catalog_admin",
+                Map.of("pr", principalRoleName, "cat", catalogName))
+            .delete()) {
+      assertThat(response).returns(Response.Status.NO_CONTENT.getStatusCode(), Response::getStatus);
+    }
+
+    // Verify catalog_role_manager is auto-revoked
+    try (Response response =
+        managementApi
+            .request(
+                "v1/principals/{p}/principal-roles",
+                Map.of("p", principal.getPrincipal().getName()))
+            .get()) {
+      assertThat(response.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+      PrincipalRoles roles = response.readEntity(PrincipalRoles.class);
+      assertThat(roles.getRoles())
+          .extracting(PrincipalRole::getName)
+          .doesNotContain(PolarisEntityConstants.getNameOfCatalogRoleManagerPrincipalRole());
+    }
+  }
+
   @Test
   public void testTableManageAccessCanGrantAndRevokeFromCatalogRoles() {
     // Create a PrincipalRole and a new catalog.

persistence/nosql/persistence/metastore-maintenance/src/test/java/org/apache/polaris/persistence/nosql/metastore/maintenance/TestCatalogMaintenance.java

@@ -245,15 +245,15 @@ public void catalogMaintenance() {
             true,
             OptionalInt.of(0),
             Optional.of(0L),
-            // 8 stale objects:
+            // 11 stale objects:
             // - 1 namespace (catalog state)
             // - 1 table (catalog state)
-            // - 2 grants (realm setup) - including 2 ACLs
+            // - 3 grants - including 3 ACLs
             // - 1 principal
-            // - 1 principal role
+            // - 2 principal roles
             // - 1 catalog role
             // - 1 catalog
-            Optional.of(10L));
+            Optional.of(13L));
 
     checkEntities("real state after grace", entities);
   }

polaris-core/src/main/java/org/apache/polaris/core/auth/AuthBootstrapUtil.java

@@ -104,6 +104,23 @@ public static PrincipalSecretsResult createPolarisPrincipalForRealm(
         rootContainer,
         PolarisPrivilege.SERVICE_MANAGE_ACCESS);
 
+    // create the catalog_role_manager principal role for catalog admins to list principal roles
+    PrincipalRoleEntity catalogRoleManagerPrincipalRole =
+        new PrincipalRoleEntity.Builder()
+            .setId(generateId(metaStoreManager, ctx))
+            .setName(PolarisEntityConstants.getNameOfCatalogRoleManagerPrincipalRole())
+            .setCreateTimestamp(System.currentTimeMillis())
+            .build();
+    metaStoreManager.createEntityIfNotExists(ctx, null, catalogRoleManagerPrincipalRole);
+
+    // grant PRINCIPAL_ROLE_LIST on the rootContainer to the catalogRoleManagerPrincipalRole
+    metaStoreManager.grantPrivilegeOnSecurableToRole(
+        ctx,
+        catalogRoleManagerPrincipalRole,
+        null,
+        rootContainer,
+        PolarisPrivilege.PRINCIPAL_ROLE_LIST);
+
     return metaStoreManager.loadPrincipalSecrets(ctx, rootPrincipal.getClientId());
   }
 

polaris-core/src/main/java/org/apache/polaris/core/entity/PolarisEntityConstants.java

@@ -43,6 +43,9 @@ public class PolarisEntityConstants {
   // the name of the principal role we create to manage the entire Polaris service
   private static final String ADMIN_PRINCIPAL_ROLE_NAME = "service_admin";
 
+  // the name of the principal role for catalog admins to list principal roles
+  private static final String CATALOG_ROLE_MANAGER_PRINCIPAL_ROLE_NAME = "catalog_role_manager";
+
   // 24 hours retention before purging. This should be a config
   private static final long RETENTION_TIME_IN_MS = 24 * 3600_000;
 
@@ -91,6 +94,10 @@ public static String getNameOfPrincipalServiceAdminRole() {
     return ADMIN_PRINCIPAL_ROLE_NAME;
   }
 
+  public static String getNameOfCatalogRoleManagerPrincipalRole() {
+    return CATALOG_ROLE_MANAGER_PRINCIPAL_ROLE_NAME;
+  }
+
   public static long getRetentionTimeInMs() {
     return RETENTION_TIME_IN_MS;
   }

polaris-core/src/testFixtures/java/org/apache/polaris/core/persistence/PolarisTestMetaStoreManager.java

@@ -1906,24 +1906,37 @@ void validateBootstrap() {
                 PageToken.readEverything())
             .getEntities();
 
-    // ensure not null, one element only
-    Assertions.assertThat(principalRoles).isNotNull().hasSize(1);
-
-    // get catalog list information
-    EntityNameLookupRecord roleListInfo = principalRoles.get(0);
+    // ensure not null, two elements (service_admin and catalog_role_manager)
+    Assertions.assertThat(principalRoles).isNotNull().hasSize(2);
+
+    // validate service_admin principal role
+    PolarisBaseEntity serviceAdminRole = null;
+    PolarisBaseEntity catalogRoleManagerRole = null;
+    for (EntityNameLookupRecord roleListInfo : principalRoles) {
+      PolarisBaseEntity role =
+          this.ensureExistsById(
+              null,
+              roleListInfo.getId(),
+              true,
+              roleListInfo.getName(),
+              PolarisEntityType.PRINCIPAL_ROLE,
+              PolarisEntitySubType.NULL_SUBTYPE);
+
+      if (role.getName().equals(PolarisEntityConstants.getNameOfPrincipalServiceAdminRole())) {
+        serviceAdminRole = role;
+      } else if (role.getName()
+          .equals(PolarisEntityConstants.getNameOfCatalogRoleManagerPrincipalRole())) {
+        catalogRoleManagerRole = role;
+      }
+    }
 
-    // now make sure this principal role was properly persisted
-    PolarisBaseEntity principalRole =
-        this.ensureExistsById(
-            null,
-            roleListInfo.getId(),
-            true,
-            PolarisEntityConstants.getNameOfPrincipalServiceAdminRole(),
-            PolarisEntityType.PRINCIPAL_ROLE,
-            PolarisEntitySubType.NULL_SUBTYPE);
+    // ensure both roles were found
+    Assertions.assertThat(serviceAdminRole).isNotNull();
+    Assertions.assertThat(catalogRoleManagerRole).isNotNull();
 
-    // also between the principal_role and the principal
-    this.ensureGrantRecordExists(principalRole, principal, PolarisPrivilege.PRINCIPAL_ROLE_USAGE);
+    // validate grant record between service_admin and root principal
+    this.ensureGrantRecordExists(
+        serviceAdminRole, principal, PolarisPrivilege.PRINCIPAL_ROLE_USAGE);
   }
 
   public void testCreateTestCatalog() {
@@ -2071,7 +2084,7 @@ void testBrowse() {
             ImmutablePair.of("P1", PolarisEntitySubType.NULL_SUBTYPE),
             ImmutablePair.of("P2", PolarisEntitySubType.NULL_SUBTYPE)));
 
-    // 3 principal roles with the bootstrap service_admin
+    // 4 principal roles with the bootstrap service_admin and catalog_role_manager
     this.validateListReturn(
         null,
         PolarisEntityType.PRINCIPAL_ROLE,
@@ -2080,6 +2093,9 @@ void testBrowse() {
             ImmutablePair.of("PR2", PolarisEntitySubType.NULL_SUBTYPE),
             ImmutablePair.of(
                 PolarisEntityConstants.getNameOfPrincipalServiceAdminRole(),
+                PolarisEntitySubType.NULL_SUBTYPE),
+            ImmutablePair.of(
+                PolarisEntityConstants.getNameOfCatalogRoleManagerPrincipalRole(),
                 PolarisEntitySubType.NULL_SUBTYPE)));
 
     // three namespaces under top-level namespace N1
@@ -2147,6 +2163,9 @@ void testBrowse() {
             ImmutablePair.of(
                 PolarisEntityConstants.getNameOfPrincipalServiceAdminRole(),
                 PolarisEntitySubType.NULL_SUBTYPE),
+            ImmutablePair.of(
+                PolarisEntityConstants.getNameOfCatalogRoleManagerPrincipalRole(),
+                PolarisEntitySubType.NULL_SUBTYPE),
             ImmutablePair.of("PR1", PolarisEntitySubType.NULL_SUBTYPE),
             ImmutablePair.of("PR2", PolarisEntitySubType.NULL_SUBTYPE)));