[WIP]HIVE-29228: Support vended credentials for S3

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

@@ -1928,6 +1928,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."
@@ -1936,10 +1940,13 @@ public enum ConfVars {
         "hive.metastore.iceberg.catalog.cache.expiry", -1,
         "HMS Iceberg Catalog cache expiry."
     ),
+    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."
-    ),
     HTTPSERVER_THREADPOOL_MIN("hive.metastore.httpserver.threadpool.min",
             "hive.metastore.httpserver.threadpool.min", 8,
             "HMS embedded HTTP server minimum number of threads."

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

@@ -26,6 +26,13 @@
     <iceberg.version>1.10.1</iceberg.version>
   </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>
@@ -72,10 +79,8 @@
     </dependency>
     <!-- Test dependencies -->
     <dependency>
-      <groupId>org.apache.hive</groupId>
-      <artifactId>hive-exec</artifactId>
-      <version>${hive.version}</version>
-      <classifier>core</classifier>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>bundle</artifactId>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -99,6 +104,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,26 @@
+/*
+ * 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;
@@ -110,16 +111,18 @@ public class HMSCatalogAdapter implements Closeable {
   private final Catalog catalog;
   private final SupportsNamespaces asNamespaceCatalog;
   private final ViewCatalog asViewCatalog;
+  private final IcebergVendedCredentialProvider credentialProvider;
   private final List<IcebergMetricsReporter> metricsReporters;
   private final Clock clock = Clock.systemUTC();
 
-  public HMSCatalogAdapter(String catalogName, Catalog catalog, List<IcebergMetricsReporter> metricsReporters) {
+  public HMSCatalogAdapter(String catalogName, Catalog catalog, 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.credentialProvider = credentialProvider;
     this.metricsReporters = metricsReporters;
   }
 
@@ -274,18 +277,21 @@ private ListTablesResponse listTables(Map<String, String> vars) {
     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()) {
-      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) {
@@ -303,21 +309,33 @@ private RESTResponse tableExists(Map<String, String> vars) {
     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) {
@@ -390,6 +408,33 @@ private RESTResponse dropView(Map<String, String> vars) {
     return null;
   }
 
+  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(ident, response.tableMetadata().location());
+    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
@@ -421,7 +466,10 @@ private static void commitTransaction(Catalog catalog, CommitTransactionRequest
 
   @SuppressWarnings({"MethodLength", "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) {
     switch (route) {
       case CONFIG:
         return (T) config();
@@ -448,7 +496,7 @@ private <T extends RESTResponse> T handleRequest(
         return (T) listTables(vars);
 
       case CREATE_TABLE:
-        return (T) createTable(vars, body);
+        return (T) createTable(accessDelegationModes, vars, body);
 
       case DROP_TABLE:
         return (T) dropTable(vars);
@@ -457,13 +505,13 @@ private <T extends RESTResponse> T handleRequest(
         return (T) tableExists(vars);
 
       case LOAD_TABLE:
-        return (T) loadTable(vars);
+        return (T) loadTable(accessDelegationModes, vars);
 
       case REGISTER_TABLE:
-        return (T) registerTable(vars, body);
+        return (T) registerTable(accessDelegationModes, vars, body);
 
       case UPDATE_TABLE:
-        return (T) updateTable(vars, body);
+        return (T) updateTable(accessDelegationModes, vars, body);
 
       case RENAME_TABLE:
         return (T) renameTable(body);
@@ -504,6 +552,7 @@ private <T extends RESTResponse> T handleRequest(
   <T extends RESTResponse> T execute(
       HTTPMethod method,
       String path,
+      Set<AccessDelegationMode> accessDelegationModes,
       Map<String, String> queryParams,
       Object body,
       HttpServletResponse response) throws IOException {
@@ -516,7 +565,7 @@ <T extends RESTResponse> T execute(
           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

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

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;
@@ -43,7 +46,7 @@ public class HMSCatalogServlet extends HttpServlet {
   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 @@ protected void service(HttpServletRequest request, HttpServletResponse response)
           restCatalogAdapter.execute(
               context.method(),
               context.path(),
+              context.accessDelegationModes(),
               context.queryParams(),
               context.body(),
               response);
@@ -95,6 +99,7 @@ public void destroy() {
   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(ErrorResponse errorResponse) {
     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 @@ static ServletRequestContext from(HttpServletRequest request) throws IOException
                 .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 @@ static ServletRequestContext from(HttpServletRequest request) throws IOException
           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 @@ public String path() {
       return path;
     }
 
+    public Set<AccessDelegationMode> accessDelegationModes() {
+      return accessDelegationModes;
+    }
+
     public Map<String, String> queryParams() {
       return queryParams;
     }