[#10684] feat(iceberg-rest): Support vended credentials on registerTable endpoint

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,21 @@ public LoadTableResponse loadTable(
     return loadTableResponse;
   }
 
+  public LoadTableResponse registerTable(
+      Namespace namespace, RegisterTableRequest request, boolean requestCredential) {
+    LoadTableResponse loadTableResponse = super.registerTable(namespace, request);
+    if (shouldGenerateCredential(loadTableResponse, requestCredential)) {
+      // Vend WRITE credentials: the registering user becomes the table owner
+      // (IcebergNamespaceHookDispatcher.setTableOwner runs after this call
+      // returns), consistent with createTable which also vends WRITE.
+      return injectCredentialConfig(
+          TableIdentifier.of(namespace, request.name()),
+          loadTableResponse,
+          CredentialPrivilege.WRITE);
+    }
+    return loadTableResponse;
+  }
+
   @Override
   public LoadTableResponse updateTable(
       TableIdentifier tableIdentifier, UpdateTableRequest updateTableRequest) {
@@ -286,7 +302,8 @@ protected boolean useDifferentClassLoader() {
     return false;
   }
 
-  private LoadTableResponse injectCredentialConfig(
+  @VisibleForTesting
+  protected LoadTableResponse injectCredentialConfig(
       TableIdentifier tableIdentifier,
       LoadTableResponse loadTableResponse,
       CredentialPrivilege privilege) {
@@ -336,7 +353,8 @@ private Credential getCredential(
     return credential;
   }
 
-  private boolean shouldGenerateCredential(
+  @VisibleForTesting
+  protected boolean shouldGenerateCredential(
       LoadTableResponse loadTableResponse, boolean requestCredential) {
     if (!requestCredential) {
       return false;

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 (WRITE).
+    if (shouldGenerateCredential(loadTableResponse, requestCredential)) {
+      return injectCredentialConfig(
+          TableIdentifier.of(namespace, request.name()),
+          loadTableResponse,
+          CredentialPrivilege.WRITE);
+    }
     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());
+  }
 }