[#10684] feat(iceberg-rest): Support credential vending on registerTable endpoint

Add X-Iceberg-Access-Delegation header support to the registerTable
endpoint, enabling credential vending in the response. Mirrors the
existing pattern from createTable and loadTable, including the
RESTCatalog delegation branch that forwards credential requests to
the upstream catalog when Gravitino fronts another REST catalog.

Author: sachinnn99

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

@@ -80,6 +80,7 @@
 import org.apache.iceberg.rest.RESTCatalog;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
 import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
@@ -156,6 +157,30 @@ public LoadTableResponse loadTable(
     return loadTableResponse;
   }
 
+  public LoadTableResponse registerTable(
+      Namespace namespace, RegisterTableRequest request, boolean requestCredential) {
+    LoadTableResponse loadTableResponse;
+    if (catalog instanceof RESTCatalog) {
+      loadTableResponse = registerTableInternal(namespace, request);
+    } else {
+      loadTableResponse = super.registerTable(namespace, request);
+    }
+    if (shouldGenerateCredential(loadTableResponse, requestCredential)) {
+      // Use READ here, not WRITE. registerTable writes no files at the storage layer,
+      // and (unlike createTable) gravitino does not make the registering user the table
+      // owner — so a user with only ANY_CREATE_TABLE privilege cannot pass
+      // FILTER_MODIFY_TABLE on a follow-up loadTable either, and would correctly receive
+      // READ there. Hardcoding WRITE here would be a one-off privilege escalation that
+      // bypasses gravitino's authz model. Higher-privileged users who need to write to a
+      // registered table must be granted MODIFY_TABLE explicitly.
+      return injectCredentialConfig(
+          TableIdentifier.of(namespace, request.name()),
+          loadTableResponse,
+          CredentialPrivilege.READ);
+    }
+    return loadTableResponse;
+  }
+
   @Override
   public LoadTableResponse updateTable(
       TableIdentifier tableIdentifier, UpdateTableRequest updateTableRequest) {
@@ -276,7 +301,16 @@ protected boolean useDifferentClassLoader() {
     return false;
   }
 
-  private LoadTableResponse injectCredentialConfig(
+  /**
+   * Injects vended credentials and catalog-client config into a {@link LoadTableResponse}.
+   *
+   * <p>Visibility is {@code protected} (rather than {@code private}) to allow subclasses to inject
+   * credentials in code paths where they cannot delegate to the matching {@code super} method —
+   * e.g. test/mock subclasses that synthesize a {@link LoadTableResponse} for operations the
+   * underlying catalog cannot perform natively (such as {@code registerTable} against an in-memory
+   * catalog).
+   */
+  protected LoadTableResponse injectCredentialConfig(
       TableIdentifier tableIdentifier,
       LoadTableResponse loadTableResponse,
       CredentialPrivilege privilege) {
@@ -326,7 +360,15 @@ private Credential getCredential(
     return credential;
   }
 
-  private boolean shouldGenerateCredential(
+  /**
+   * Decides whether to vend credentials for a given {@link LoadTableResponse}.
+   *
+   * <p>Visibility is {@code protected} (rather than {@code private}) for the same reason as {@link
+   * #injectCredentialConfig}: subclasses that build a {@link LoadTableResponse} without calling
+   * {@code super} (e.g. test wrappers around in-memory catalogs) need to honor the same vending
+   * gate that production paths use.
+   */
+  protected boolean shouldGenerateCredential(
       LoadTableResponse loadTableResponse, boolean requestCredential) {
     if (!requestCredential) {
       return false;
@@ -726,6 +768,26 @@ private LoadTableResponse loadTableInternal(TableIdentifier ident) {
     throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable");
   }
 
+  private LoadTableResponse registerTableInternal(
+      Namespace namespace, RegisterTableRequest request) {
+    Table table =
+        catalog.registerTable(
+            TableIdentifier.of(namespace, request.name()), request.metadataLocation());
+
+    if (table instanceof BaseTable) {
+      Map<String, String> properties = retrieveFileIOProperties(table.io());
+      return LoadTableResponse.builder()
+          .withTableMetadata(((BaseTable) table).operations().current())
+          .addAllConfig(
+              MapUtils.getFilteredMap(
+                  properties, key -> catalogPropertiesToClientKeys.contains(key)))
+          .addAllConfig(CredentialPropertyUtils.filterCredentialProperties(properties))
+          .build();
+    }
+
+    throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable");
+  }
+
   private static Map<String, String> retrieveFileIOProperties(FileIO fileIO) {
     return fileIO instanceof InMemoryFileIO ? Maps.newHashMap() : fileIO.properties();
   }

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

@@ -123,6 +123,6 @@ public LoadTableResponse registerTable(
       RegisterTableRequest registerTableRequest) {
     return icebergCatalogWrapperManager
         .getCatalogWrapper(context.catalogName())
-        .registerTable(namespace, registerTableRequest);
+        .registerTable(namespace, registerTableRequest, context.requestCredentialVending());
   }
 }

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

@@ -33,6 +33,7 @@
 import javax.ws.rs.Encoded;
 import javax.ws.rs.GET;
 import javax.ws.rs.HEAD;
+import javax.ws.rs.HeaderParam;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
@@ -300,20 +301,25 @@ public Response registerTable(
       @AuthorizationMetadata(type = Entity.EntityType.CATALOG) @PathParam("prefix") String prefix,
       @AuthorizationMetadata(type = Entity.EntityType.SCHEMA) @Encoded() @PathParam("namespace")
           String namespace,
-      RegisterTableRequest registerTableRequest) {
+      RegisterTableRequest registerTableRequest,
+      @HeaderParam(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION) String accessDelegation) {
+    boolean isCredentialVending = IcebergTableOperations.isCredentialVending(accessDelegation);
     String catalogName = IcebergRESTUtils.getCatalogName(prefix);
     Namespace icebergNS = RESTUtil.decodeNamespace(namespace);
     LOG.info(
-        "Register Iceberg table, catalog: {}, namespace: {}, registerTableRequest: {}",
+        "Register Iceberg table, catalog: {}, namespace: {}, registerTableRequest: {}, "
+            + "accessDelegation: {}, isCredentialVending: {}",
         catalogName,
         icebergNS,
-        registerTableRequest);
+        registerTableRequest,
+        accessDelegation,
+        isCredentialVending);
     try {
       return Utils.doAs(
           httpRequest,
           () -> {
             IcebergRequestContext context =
-                new IcebergRequestContext(httpServletRequest(), catalogName);
+                new IcebergRequestContext(httpServletRequest(), catalogName, isCredentialVending);
             LoadTableResponse loadTableResponse =
                 namespaceOperationDispatcher.registerTable(
                     context, icebergNS, registerTableRequest);

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

@@ -603,7 +603,13 @@ private static boolean etagMatches(String ifNoneMatch, EntityTag etag) {
     return etag.getValue().equals(clientEtag);
   }
 
-  private boolean isCredentialVending(String accessDelegation) {
+  /**
+   * Parses the {@code X-Iceberg-Access-Delegation} header value and returns whether the client is
+   * requesting credential vending. Package-private and static so that {@link
+   * IcebergNamespaceOperations#registerTable} can reuse the same parsing logic from the same
+   * package.
+   */
+  static boolean isCredentialVending(String accessDelegation) {
     if (StringUtils.isBlank(accessDelegation)) {
       return false;
     }

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/CatalogWrapperForTest.java

@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import org.apache.gravitino.credential.CredentialPrivilege;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.iceberg.service.CatalogWrapperForREST;
 import org.apache.iceberg.DataFile;
@@ -41,7 +42,9 @@
 import org.apache.iceberg.types.Types.NestedField;
 import org.apache.iceberg.types.Types.StringType;
 
-// Used to override registerTable
+// Test wrapper that mocks operations the in-memory catalog cannot perform natively
+// (e.g. registerTable, which requires a real metadata.json file at the given location),
+// and adds test-only hooks (e.g. plan-task data generation for createTable).
 @SuppressWarnings("deprecation")
 public class CatalogWrapperForTest extends CatalogWrapperForREST {
   public static final String GENERATE_PLAN_TASKS_DATA_PROP = "test.generate-plan-data";
@@ -61,23 +64,44 @@ public LoadTableResponse createTable(
   }
 
   @Override
-  public LoadTableResponse registerTable(Namespace namespace, RegisterTableRequest request) {
+  public LoadTableResponse registerTable(
+      Namespace namespace, RegisterTableRequest request, boolean requestCredential) {
     if (request.name().contains("fail")) {
       throw new AlreadyExistsException("Already exits exception for test");
     }
 
+    // The in-memory test catalog cannot natively registerTable (it would need a real
+    // metadata.json file at the given location), so build a mock LoadTableResponse here.
+    // Honor cloud URIs (e.g. s3://) in metadataLocation so credential vending tests can
+    // verify the vended path; default to /mock otherwise for existing tests.
+    String location =
+        request.metadataLocation().contains("://") ? request.metadataLocation() : "/mock";
     Schema mockSchema = new Schema(NestedField.of(1, false, "foo_string", StringType.get()));
     TableMetadata baseMetadata =
         TableMetadata.newTableMetadata(
-            mockSchema, PartitionSpec.unpartitioned(), "/mock", ImmutableMap.of());
+            mockSchema, PartitionSpec.unpartitioned(), location, ImmutableMap.of());
     String json = TableMetadataParser.toJson(baseMetadata);
     TableMetadata tableMetadata =
-        TableMetadataParser.fromJson("/mock/metadata/v1.metadata.json", json);
+        TableMetadataParser.fromJson(location + "/metadata/v1.metadata.json", json);
     LoadTableResponse loadTableResponse =
         LoadTableResponse.builder()
             .withTableMetadata(tableMetadata)
             .addAllConfig(ImmutableMap.of())
             .build();
+    // We must replicate the credential-vending check + injection here (rather than reuse
+    // CatalogWrapperForREST.registerTable via super) because the in-memory test catalog
+    // cannot natively perform registerTable. Above, we synthesized a mock LoadTableResponse
+    // instead of delegating to super.registerTable; that means the parent class never sees
+    // this call and therefore never runs its vending logic. Calling shouldGenerateCredential
+    // / injectCredentialConfig directly (now protected on the parent) re-applies the exact
+    // same vending behavior to the mock response, so credential-vending tests exercise the
+    // production code path end-to-end. Privilege must match the production wrapper (READ).
+    if (shouldGenerateCredential(loadTableResponse, requestCredential)) {
+      return injectCredentialConfig(
+          TableIdentifier.of(namespace, request.name()),
+          loadTableResponse,
+          CredentialPrivilege.READ);
+    }
     return loadTableResponse;
   }
 

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergNamespaceTestBase.java

@@ -59,6 +59,18 @@ protected Response doRegisterTable(String tableName, Namespace ns) {
         .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
   }
 
+  protected Response doRegisterTableWithCredentialVending(
+      String tableName, Namespace ns, String metadataLocation) {
+    RegisterTableRequest request =
+        ImmutableRegisterTableRequest.builder()
+            .name(tableName)
+            .metadataLocation(metadataLocation)
+            .build();
+    return getNamespaceClientBuilder(Optional.of(ns), Optional.of("register"), Optional.empty())
+        .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION, "vended-credentials")
+        .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+  }
+
   private Response doListNamespace(Optional<Namespace> parent) {
     Optional<Map<String, String>> queryParam =
         parent.isPresent()

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergNamespaceOperations.java

@@ -21,10 +21,15 @@
 import java.util.Arrays;
 import java.util.Optional;
 import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.client.Entity;
 import javax.ws.rs.core.Application;
 import javax.ws.rs.core.EntityTag;
+import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import org.apache.gravitino.credential.Credential;
 import org.apache.gravitino.iceberg.service.IcebergRESTUtils;
+import org.apache.gravitino.iceberg.service.extension.DummyCredentialProvider;
 import org.apache.gravitino.listener.api.event.Event;
 import org.apache.gravitino.listener.api.event.IcebergCreateNamespaceEvent;
 import org.apache.gravitino.listener.api.event.IcebergCreateNamespaceFailureEvent;
@@ -44,6 +49,9 @@
 import org.apache.gravitino.listener.api.event.IcebergUpdateNamespaceFailureEvent;
 import org.apache.gravitino.listener.api.event.IcebergUpdateNamespacePreEvent;
 import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.rest.requests.ImmutableRegisterTableRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
 import org.glassfish.jersey.internal.inject.AbstractBinder;
 import org.glassfish.jersey.server.ResourceConfig;
 import org.junit.jupiter.api.Assertions;
@@ -257,4 +265,68 @@ void testUpdateNamespace() {
     dummyEventListener.clearEvent();
     verifyUpdateNamespaceSucc(Namespace.of("update_foo3", "a"));
   }
+
+  @Test
+  void testRegisterTableWithCredentialVending() {
+    // register without credential vending — no credentials in response
+    verifyRegisterTableSucc("register_cred_foo1", Namespace.of("register_cred_ns"));
+
+    // register with credential vending but local location — should NOT vend
+    Response response =
+        doRegisterTableWithCredentialVending(
+            "register_cred_foo2", Namespace.of("register_cred_ns2"), "mock");
+    Assertions.assertEquals(Status.OK.getStatusCode(), response.getStatus());
+    LoadTableResponse loadTableResponse = response.readEntity(LoadTableResponse.class);
+    Assertions.assertFalse(loadTableResponse.config().containsKey(Credential.CREDENTIAL_TYPE));
+
+    // register with credential vending and S3 location — SHOULD vend
+    String s3Location = "s3://dummy-bucket/register_cred_foo3";
+    response =
+        doRegisterTableWithCredentialVending(
+            "register_cred_foo3", Namespace.of("register_cred_ns3"), s3Location);
+    Assertions.assertEquals(Status.OK.getStatusCode(), response.getStatus());
+    loadTableResponse = response.readEntity(LoadTableResponse.class);
+    Assertions.assertEquals(
+        DummyCredentialProvider.DUMMY_CREDENTIAL_TYPE,
+        loadTableResponse.config().get(Credential.CREDENTIAL_TYPE));
+    // DummyCredentialProvider.SimpleCredential is not one of the typed credentials handled
+    // in CredentialPropertyUtils#toIcebergProperties, so it falls through to
+    // Credential#toProperties, which always emits credential-type and expire-time-in-ms.
+    // Asserting both guards against partial-injection regressions.
+    Assertions.assertEquals("0", loadTableResponse.config().get(Credential.EXPIRE_TIME_IN_MS));
+  }
+
+  @Test
+  void testRegisterTableRemoteSigningNotSupported() {
+    RegisterTableRequest request =
+        ImmutableRegisterTableRequest.builder()
+            .name("remote_signing_test")
+            .metadataLocation("mock")
+            .build();
+    Response response =
+        getNamespaceClientBuilder(
+                Optional.of(Namespace.of("register_remote_ns")),
+                Optional.of("register"),
+                Optional.empty())
+            .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION, "remote-signing")
+            .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+    Assertions.assertEquals(406, response.getStatus());
+  }
+
+  @Test
+  void testRegisterTableInvalidAccessDelegation() {
+    RegisterTableRequest request =
+        ImmutableRegisterTableRequest.builder()
+            .name("invalid_delegation_test")
+            .metadataLocation("mock")
+            .build();
+    Response response =
+        getNamespaceClientBuilder(
+                Optional.of(Namespace.of("register_invalid_ns")),
+                Optional.of("register"),
+                Optional.empty())
+            .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION, "invalid-value")
+            .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+    Assertions.assertEquals(400, response.getStatus());
+  }
 }