[WIP]HIVE-29228: Support vended credentials in Iceberg REST API server

standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java

@@ -1970,6 +1970,10 @@ public enum ConfVars {
         "The pattern to extract a user name. This is effective when you use RegexPrincipalMapper. For example, if " +
         "you want to extract a user name from the local part of the email claim, set this to (.*)@example.com."
     ),
+    CATALOG_VENDED_CREDENTIALS_PROVIDERS("metastore.catalog.vended-credentials.providers",
+        "hive.metastore.catalog.vended-credentials.providers", "",
+        "List of comma-separated credential-vending provider IDs"
+    ),
     ICEBERG_CATALOG_SERVLET_PATH("metastore.iceberg.catalog.servlet.path",
         "hive.metastore.iceberg.catalog.servlet.path", "iceberg",
         "HMS Iceberg Catalog servlet path component of URL endpoint."
@@ -1982,6 +1986,10 @@ public enum ConfVars {
         "hive.metastore.iceberg.catalog.unique.table.location", false,
         "Whether the HMS Iceberg REST catalog should assign a unique storage location for each new table."
     ),
+    ICEBERG_CATALOG_VENDED_CREDENTIALS_ENABLED("metastore.iceberg.catalog.vended-credentials.enabled",
+        "hive.metastore.iceberg.catalog.vended-credentials.enabled", false,
+        "Boolean flag to enable credential vending on Iceberg REST Catalog"
+    ),
     ICEBERG_CATALOG_METRICS_REPORTERS("metastore.iceberg.catalog.metrics.reporters",
         "hive.metastore.iceberg.catalog.metrics.reporters", "org.apache.iceberg.rest.metrics.LoggingMetricsReporter",
         "A comma separated list of custom Iceberg Metrics Reporting plugins."

standalone-metastore/metastore-rest-catalog/pom.xml

@@ -31,6 +31,13 @@
     <log4j2.debug>false</log4j2.debug>
   </properties>
   <dependencies>
+    <dependency>
+      <groupId>org.apache.hive</groupId>
+      <artifactId>hive-exec</artifactId>
+      <version>${hive.version}</version>
+      <classifier>core</classifier>
+      <scope>provided</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.hive</groupId>
       <artifactId>hive-standalone-metastore-server</artifactId>
@@ -84,6 +91,11 @@
       <version>1.9.17</version>
     </dependency>
     <!-- Test dependencies -->
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>bundle</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.hive</groupId>
       <artifactId>hive-standalone-metastore-common</artifactId>
@@ -105,6 +117,17 @@
       <classifier>tests</classifier>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.iceberg</groupId>
+      <artifactId>iceberg-aws</artifactId>
+      <version>${iceberg.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-aws</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.iceberg</groupId>
       <artifactId>iceberg-core</artifactId>

standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/AccessDelegationMode.java

@@ -0,0 +1,27 @@
+/*
+ * 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.iceberg.rest;
+
+/**
+ * Possible values for the X-Iceberg-Access-Delegation header.
+ */
+public enum AccessDelegationMode {
+  VENDED_CREDENTIALS, REMOTE_SIGNING
+}

standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogAdapter.java

@@ -27,6 +27,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import javax.servlet.http.HttpServletResponse;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.BaseTransaction;
@@ -82,7 +83,7 @@
 import org.slf4j.LoggerFactory;
 
 /**
  * Original @ <a href="https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java">RESTCatalogAdapter.java</a>
  * Adaptor class to translate REST requests into {@link Catalog} API calls.
  */
 public class HMSCatalogAdapter implements Closeable {
@@ -120,18 +121,20 @@
   private final SupportsNamespaces asNamespaceCatalog;
   private final ViewCatalog asViewCatalog;
   private final IcebergAuthorizer icebergAuthorizer;
+  private final IcebergVendedCredentialProvider credentialProvider;
   private final List<IcebergMetricsReporter> metricsReporters;
   private final Clock clock = Clock.systemUTC();
 
-  public HMSCatalogAdapter(String catalogName, Catalog catalog, IcebergAuthorizer icebergAuthorizer,
-      List<IcebergMetricsReporter> metricsReporters) {
+  HMSCatalogAdapter(String catalogName, Catalog catalog, IcebergAuthorizer icebergAuthorizer,
+        IcebergVendedCredentialProvider credentialProvider, List<IcebergMetricsReporter> metricsReporters) {
     Preconditions.checkArgument(catalog instanceof SupportsNamespaces);
     Preconditions.checkArgument(catalog instanceof ViewCatalog);
     this.catalogName = catalogName;
     this.catalog = catalog;
     this.asNamespaceCatalog = (SupportsNamespaces) catalog;
     this.asViewCatalog = (ViewCatalog) catalog;
     this.icebergAuthorizer = icebergAuthorizer;
+    this.credentialProvider = credentialProvider;
     this.metricsReporters = metricsReporters;
   }
 
@@ -239,7 +242,7 @@
       int prefixLength = prefixLength(requestPath);
 
       ImmutableMap.Builder<String, String> vars = ImmutableMap.builder();
       for (Map.Entry<Integer, String> var : variables.entrySet()) {
         vars.put(var.getValue(), requestPath.get(mappedIndex(var.getKey(), prefixLength)));
       }
 
@@ -325,20 +328,23 @@
     return castResponse(ListTablesResponse.class, CatalogHandlers.listTables(catalog, namespace));
   }
 
-  private LoadTableResponse createTable(Map<String, String> vars, Object body) {
+  private LoadTableResponse createTable(Set<AccessDelegationMode> accessDelegationModes, Map<String, String> vars,
+      Object body) {
     final Class<LoadTableResponse> responseType = LoadTableResponse.class;
     Namespace namespace = namespaceFromPathVars(vars);
     CreateTableRequest request = castRequest(CreateTableRequest.class, body);
     request.validate();
+    LoadTableResponse response;
     if (request.stageCreate()) {
       Map<String, String> namespaceMetadata = asNamespaceCatalog.loadNamespaceMetadata(namespace);
       icebergAuthorizer.validateStageCreateTable(catalogName, namespace, namespaceMetadata, request);
-      return castResponse(
-              responseType, CatalogHandlers.stageTableCreate(catalog, namespace, request));
+      response = castResponse(
+          responseType, CatalogHandlers.stageTableCreate(catalog, namespace, request));
     } else {
-      return castResponse(
-              responseType, CatalogHandlers.createTable(catalog, namespace, request));
+      response = castResponse(
+          responseType, CatalogHandlers.createTable(catalog, namespace, request));
     }
+    return withCredentials(accessDelegationModes, TableIdentifier.of(namespace, request.name()), response);
   }
 
   private RESTResponse dropTable(Map<String, String> vars) {
@@ -356,21 +362,33 @@
     return null;
   }
 
-  private LoadTableResponse loadTable(Map<String, String> vars) {
+  private LoadTableResponse loadTable(Set<AccessDelegationMode> delegationModes, Map<String, String> vars) {
     TableIdentifier ident = identFromPathVars(vars);
-    return castResponse(LoadTableResponse.class, CatalogHandlers.loadTable(catalog, ident));
+    LoadTableResponse response =
+        castResponse(LoadTableResponse.class, CatalogHandlers.loadTable(catalog, ident));
+    return withCredentials(delegationModes, ident, response);
   }
 
-  private LoadTableResponse registerTable(Map<String, String> vars, Object body) {
-      Namespace namespace = namespaceFromPathVars(vars);
-      RegisterTableRequest request = castRequest(RegisterTableRequest.class, body);
-      return castResponse(LoadTableResponse.class, CatalogHandlers.registerTable(catalog, namespace, request));
+  private LoadTableResponse registerTable(
+      Set<AccessDelegationMode> delegationModes,
+      Map<String, String> vars,
+      Object body) {
+    Namespace namespace = namespaceFromPathVars(vars);
+    RegisterTableRequest request = castRequest(RegisterTableRequest.class, body);
+    LoadTableResponse response =
+        castResponse(LoadTableResponse.class, CatalogHandlers.registerTable(catalog, namespace, request));
+    return withCredentials(delegationModes, TableIdentifier.of(namespace, request.name()), response);
   }
 
-  private LoadTableResponse updateTable(Map<String, String> vars, Object body) {
+  private LoadTableResponse updateTable(
+      Set<AccessDelegationMode> delegationModes,
+      Map<String, String> vars,
+      Object body) {
     TableIdentifier ident = identFromPathVars(vars);
     UpdateTableRequest request = castRequest(UpdateTableRequest.class, body);
-    return castResponse(LoadTableResponse.class, CatalogHandlers.updateTable(catalog, ident, request));
+    LoadTableResponse response =
+        castResponse(LoadTableResponse.class, CatalogHandlers.updateTable(catalog, ident, request));
+    return withCredentials(delegationModes, ident, response);
   }
 
   private RESTResponse renameTable(Object body) {
@@ -450,6 +468,33 @@
         LoadViewResponse.class, CatalogHandlers.registerView(asViewCatalog, namespace, request));
   }
 
+  private LoadTableResponse withCredentials(
+      Set<AccessDelegationMode> accessDelegationModes,
+      TableIdentifier ident,
+      LoadTableResponse response) {
+    if (credentialProvider == null) {
+      return response;
+    }
+
+    if (accessDelegationModes.contains(AccessDelegationMode.VENDED_CREDENTIALS)) {
+      return withVendedCredentials(ident, response);
+    }
+
+    if (accessDelegationModes.contains(AccessDelegationMode.REMOTE_SIGNING)) {
+      LOG.warn("Remote signing is not supported. Ignoring...");
+    }
+
+    return response;
+  }
+
+  private LoadTableResponse withVendedCredentials(TableIdentifier ident, LoadTableResponse response) {
+    final var credentials = credentialProvider.vend(catalogName, ident, response.tableMetadata());
+    return LoadTableResponse.builder()
+        .withTableMetadata(response.tableMetadata())
+        .addAllConfig(response.config())
+        .addAllCredentials(credentials).build();
+  }
+
   /**
    * This is a very simplistic approach that only validates the requirements for each table and does
    * not do any other conflict detection. Therefore, it does not guarantee true transactional
@@ -481,39 +526,43 @@
 
   @SuppressWarnings({"unchecked"})
   private <T extends RESTResponse> T handleRequest(
-      Route route, Map<String, String> vars, Object body) {
+      Route route,
+      Set<AccessDelegationMode> accessDelegationModes,
+      Map<String, String> vars,
+      Object body) {
     return (T) switch (route) {
       case CONFIG -> config();
       case LIST_NAMESPACES -> listNamespaces(vars);
       case CREATE_NAMESPACE -> createNamespace(body);
       case NAMESPACE_EXISTS -> namespaceExists(vars);
       case LOAD_NAMESPACE -> loadNamespace(vars);
       case DROP_NAMESPACE -> dropNamespace(vars);
       case UPDATE_NAMESPACE -> updateNamespace(vars, body);
       case LIST_TABLES -> listTables(vars);
-      case CREATE_TABLE -> createTable(vars, body);
+      case CREATE_TABLE -> createTable(accessDelegationModes, vars, body);
       case DROP_TABLE -> dropTable(vars);
       case TABLE_EXISTS -> tableExists(vars);
-      case LOAD_TABLE -> loadTable(vars);
-      case REGISTER_TABLE -> registerTable(vars, body);
-      case UPDATE_TABLE -> updateTable(vars, body);
+      case LOAD_TABLE -> loadTable(accessDelegationModes, vars);
+      case REGISTER_TABLE -> registerTable(accessDelegationModes, vars, body);
+      case UPDATE_TABLE -> updateTable(accessDelegationModes, vars, body);
       case RENAME_TABLE -> renameTable(body);
       case REPORT_METRICS -> reportMetrics(vars, body);
       case COMMIT_TRANSACTION -> commitTransaction(body);
       case LIST_VIEWS -> listViews(vars);
       case CREATE_VIEW -> createView(vars, body);
       case VIEW_EXISTS -> viewExists(vars);
       case LOAD_VIEW -> loadView(vars);
       case UPDATE_VIEW -> updateView(vars, body);
       case RENAME_VIEW -> renameView(body);
       case DROP_VIEW -> dropView(vars);
       case REGISTER_VIEW -> registerView(vars, body);
     };
   }
 
   <T extends RESTResponse> T execute(
       HTTPMethod method,
       String path,
+      Set<AccessDelegationMode> accessDelegationModes,
       Map<String, String> queryParams,
       Object body,
       HttpServletResponse response) throws IOException {
@@ -526,7 +575,7 @@
           vars.putAll(queryParams);
         }
         vars.putAll(routeAndVars.second());
-        return handleRequest(routeAndVars.first(), vars.build(), body);
+        return handleRequest(routeAndVars.first(), accessDelegationModes, vars.build(), body);
       } catch (RuntimeException e) {
         configureResponseFromException(e, errorBuilder);
       }

standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogFactory.java

@@ -117,8 +117,12 @@ private HttpServlet createServlet(Catalog catalog) {
     ServletSecurity security = new ServletSecurity(AuthType.fromString(authType), configuration, req -> scopes);
     String catalogName = MetastoreConf.getVar(configuration, ConfVars.CATALOG_DEFAULT);
     IcebergAuthorizer icebergAuthorizer = new IcebergAuthorizer(configuration);
+    IcebergVendedCredentialProvider vendedCredentialProvider = null;
+    if (MetastoreConf.getBoolVar(configuration, ConfVars.ICEBERG_CATALOG_VENDED_CREDENTIALS_ENABLED)) {
+      vendedCredentialProvider = new IcebergVendedCredentialProvider(icebergAuthorizer, configuration);
+    }
     List<IcebergMetricsReporter> reporters = createReporters();
-    var adapter = new HMSCatalogAdapter(catalogName, catalog, icebergAuthorizer, reporters);
+    var adapter = new HMSCatalogAdapter(catalogName, catalog, icebergAuthorizer, vendedCredentialProvider, reporters);
     return security.proxy(new HMSCatalogServlet(adapter));
   }
 

standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogServlet.java

@@ -20,8 +20,11 @@
 package org.apache.iceberg.rest;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.stream.Collectors;
 import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServletRequest;
@@ -35,7 +38,7 @@
 import org.slf4j.LoggerFactory;
 
 /**
  * Original @ <a href="https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/test/java/org/apache/iceberg/rest/RESTCatalogServlet.java">RESTCatalogServlet.java</a>
  * The RESTCatalogServlet provides a servlet implementation used in combination with a
  * RESTCatalogAdaptor to proxy the REST Spec to any Catalog implementation.
  */
@@ -43,7 +46,7 @@
   private static final Logger LOG = LoggerFactory.getLogger(HMSCatalogServlet.class);
   private static final String CONTENT_TYPE = "Content-Type";
   private static final String APPLICATION_JSON = "application/json";
-  
+
   private final HMSCatalogAdapter restCatalogAdapter;
   private final Map<String, String> responseHeaders =
       ImmutableMap.of(CONTENT_TYPE, APPLICATION_JSON);
@@ -73,6 +76,7 @@
           restCatalogAdapter.execute(
               context.method(),
               context.path(),
+              context.accessDelegationModes(),
               context.queryParams(),
               context.body(),
               response);
@@ -95,6 +99,7 @@
   public static class ServletRequestContext {
     private HTTPMethod method;
     private String path;
+    private Set<AccessDelegationMode> accessDelegationModes;
     private Map<String, String> queryParams;
     private Object body;
 
@@ -107,10 +112,12 @@
     private ServletRequestContext(
         HTTPMethod method,
         String path,
+        Set<AccessDelegationMode> accessDelegationModes,
         Map<String, String> queryParams,
         Object body) {
       this.method = method;
       this.path = path;
+      this.accessDelegationModes = accessDelegationModes;
       this.queryParams = queryParams;
       this.body = body;
     }
@@ -136,6 +143,21 @@
                 .build());
       }
 
+      var accessDelegationModes = Arrays
+          .stream(Optional.ofNullable(request.getHeader("X-Iceberg-Access-Delegation")).orElse("").split(","))
+          .map(String::trim)
+          .filter(header -> !header.isEmpty())
+          .map(header -> switch (header) {
+            case "vended-credentials" -> AccessDelegationMode.VENDED_CREDENTIALS;
+            case "remote-signing" -> AccessDelegationMode.REMOTE_SIGNING;
+            default -> {
+              LOG.warn("Unknown access delegation mode: {}", header);
+              yield null;
+            }
+          })
+          .filter(Objects::nonNull)
+          .collect(Collectors.toUnmodifiableSet());
+
       Route route = routeContext.first();
       Object requestBody = null;
       if (route.requestClass() != null) {
@@ -154,7 +176,7 @@
           request.getParameterMap().entrySet().stream()
               .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue()[0]));
 
-      return new ServletRequestContext(method, path, queryParams, requestBody);
+      return new ServletRequestContext(method, path, accessDelegationModes, queryParams, requestBody);
     }
 
     HTTPMethod method() {
@@ -165,6 +187,10 @@
       return path;
     }
 
+    public Set<AccessDelegationMode> accessDelegationModes() {
+      return accessDelegationModes;
+    }
+
     public Map<String, String> queryParams() {
       return queryParams;
     }