[#11968] test(iceberg): cover role narrowing in Iceberg credential vending

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergAuthorizationIT.java

@@ -37,6 +37,7 @@
 import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.MetadataObjects;
 import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.auth.AuthConstants;
 import org.apache.gravitino.authorization.Owner;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
 import org.apache.gravitino.client.GravitinoMetalake;
@@ -76,6 +77,7 @@ public class IcebergAuthorizationIT extends BaseIT {
   protected static final String METALAKE_NAME = "test_metalake";
   protected static final String GRAVITINO_CATALOG_NAME = "iceberg";
   protected static final String SPARK_CATALOG_NAME = "rest";
+  protected static final String NARROWED_SPARK_CATALOG_NAME = "narrowed";
 
   protected static final String SUPER_USER = "super";
   protected static final String NORMAL_USER = "normal";
@@ -209,6 +211,14 @@ protected boolean supportsCredentialVending() {
     return false;
   }
 
+  /**
+   * Roles sent as {@code X-Gravitino-Active-Roles} by the {@link #NARROWED_SPARK_CATALOG_NAME}
+   * catalog. Returning null registers no such catalog.
+   */
+  protected String narrowedCatalogActiveRoles() {
+    return null;
+  }
+
   void revokeUserRoles() {
     List<String> roles = metalakeClientWithAllPrivilege.getUser(NORMAL_USER).roles();
     if (roles.size() > 0) {
@@ -292,24 +302,44 @@ private void initSparkEnv() {
             .set(
                 "spark.sql.extensions",
                 "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
-            .set("spark.sql.catalog.rest", "org.apache.iceberg.spark.SparkCatalog")
-            .set("spark.sql.catalog.rest.type", "rest")
-            .set("spark.sql.catalog.rest.uri", icebergRESTUri)
-            // disable spark side table cache to check the privilege in each operation
-            .set("spark.sql.catalog.rest." + CatalogProperties.CACHE_ENABLED, "false")
-            .set("spark.sql.catalog.rest.rest.auth.type", "basic")
-            .set("spark.sql.catalog.rest.rest.auth.basic.username", NORMAL_USER)
-            .set("spark.sql.catalog.rest.rest.auth.basic.password", "mock")
             // drop Iceberg table purge may hang in spark local mode
             .set("spark.locality.wait.node", "0");
-    if (supportsCredentialVending()) {
+    configureRestCatalog(sparkConf, SPARK_CATALOG_NAME, icebergRESTUri);
+
+    String activeRoles = narrowedCatalogActiveRoles();
+    if (activeRoles != null) {
+      configureRestCatalog(sparkConf, NARROWED_SPARK_CATALOG_NAME, icebergRESTUri);
       sparkConf.set(
-          "spark.sql.catalog.rest.header.X-Iceberg-Access-Delegation", "vended-credentials");
+          sparkCatalogPrefix(NARROWED_SPARK_CATALOG_NAME)
+              + ".header."
+              + AuthConstants.X_GRAVITINO_ACTIVE_ROLES_HEADER,
+          activeRoles);
     }
 
     sparkSession = SparkSession.builder().master("local[1]").config(sparkConf).getOrCreate();
   }
 
+  private static String sparkCatalogPrefix(String catalogName) {
+    return "spark.sql.catalog." + catalogName;
+  }
+
+  private void configureRestCatalog(
+      SparkConf sparkConf, String catalogName, String icebergRESTUri) {
+    String prefix = sparkCatalogPrefix(catalogName);
+    sparkConf
+        .set(prefix, "org.apache.iceberg.spark.SparkCatalog")
+        .set(prefix + ".type", "rest")
+        .set(prefix + ".uri", icebergRESTUri)
+        // disable spark side table cache to check the privilege in each operation
+        .set(prefix + "." + CatalogProperties.CACHE_ENABLED, "false")
+        .set(prefix + ".rest.auth.type", "basic")
+        .set(prefix + ".rest.auth.basic.username", NORMAL_USER)
+        .set(prefix + ".rest.auth.basic.password", "mock");
+    if (supportsCredentialVending()) {
+      sparkConf.set(prefix + ".header.X-Iceberg-Access-Delegation", "vended-credentials");
+    }
+  }
+
   private String getPGUri() {
     return containerSuite.getPostgreSQLContainer().getJdbcUrl(TestDatabaseName.PG_ICEBERG_AUTHZ_IT);
   }

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTCloudTokenAuthorizationBaseIT.java

@@ -54,6 +54,12 @@
 public abstract class IcebergRESTCloudTokenAuthorizationBaseIT extends IcebergAuthorizationIT {
 
   protected static final String SCHEMA_NAME = "schema";
+  private static final String NARROWED_ROLE_NAME = "narrowed_select_role";
+
+  @Override
+  protected String narrowedCatalogActiveRoles() {
+    return NARROWED_ROLE_NAME;
+  }
 
   @BeforeEach
   void revokePrivilege() {
@@ -214,6 +220,50 @@ void testIcebergModifyTableCloudToken() {
     Assertions.assertEquals(2, rows.size());
   }
 
+  @Test
+  void testActiveRolesNarrowCloudToken() {
+    String tableName = "test_narrowed_" + getCloudProviderName();
+    createTable(SCHEMA_NAME, tableName);
+
+    grantNarrowedSelectRole();
+    grantModifyTableRole(tableName);
+
+    // Every held role is active, so the vended credential can write.
+    sql("INSERT INTO %s VALUES (1,1),(2,2)", tableName);
+
+    // Narrowed to the select-only role the caller gets a read-only credential, so the write fails
+    // inside Spark instead of being rejected by Gravitino.
+    Assertions.assertThrows(
+        SparkException.class,
+        () ->
+            sql(
+                "INSERT INTO %s.%s.%s VALUES (3,3)",
+                NARROWED_SPARK_CATALOG_NAME, SCHEMA_NAME, tableName));
+
+    List<Object[]> rows =
+        sql("SELECT * FROM %s.%s.%s", NARROWED_SPARK_CATALOG_NAME, SCHEMA_NAME, tableName);
+    Assertions.assertEquals(2, rows.size());
+  }
+
+  /**
+   * The narrowed catalog names this role in a static header, so it has to carry every privilege the
+   * read path needs; narrowing deactivates the per-test USE_SCHEMA role.
+   */
+  private void grantNarrowedSelectRole() {
+    SecurableObject catalogObject =
+        SecurableObjects.ofCatalog(
+            GRAVITINO_CATALOG_NAME, ImmutableList.of(Privileges.UseCatalog.allow()));
+    SecurableObject schemaObject =
+        SecurableObjects.ofSchema(
+            catalogObject,
+            SCHEMA_NAME,
+            ImmutableList.of(Privileges.UseSchema.allow(), Privileges.SelectTable.allow()));
+    metalakeClientWithAllPrivilege.createRole(
+        NARROWED_ROLE_NAME, new HashMap<>(), ImmutableList.of(catalogObject, schemaObject));
+    metalakeClientWithAllPrivilege.grantRolesToUser(
+        ImmutableList.of(NARROWED_ROLE_NAME), NORMAL_USER);
+  }
+
   protected void grantUseSchemaRole(String schema) {
     String roleName = "useSchema_" + UUID.randomUUID();
     List<SecurableObject> securableObjects = new ArrayList<>();