Add support for S3 request signing

Author: adutra

LICENSE

@@ -216,6 +216,7 @@ License: https://www.apache.org/licenses/LICENSE-2.0
 This product includes code from Apache Iceberg.
 
 * spec/iceberg-rest-catalog-open-api.yaml
+* spec/iceberg-s3-signer-open-api.yaml
 * spec/polaris-catalog-apis/oauth-tokens-api.yaml
 * integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java
 * runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java

api/iceberg-service/build.gradle.kts

@@ -31,6 +31,7 @@ dependencies {
   implementation(platform(libs.iceberg.bom))
   implementation("org.apache.iceberg:iceberg-api")
   implementation("org.apache.iceberg:iceberg-core")
+  implementation("org.apache.iceberg:iceberg-aws")
 
   implementation(libs.jakarta.annotation.api)
   implementation(libs.jakarta.inject.api)
@@ -49,6 +50,9 @@ dependencies {
   implementation("com.fasterxml.jackson.core:jackson-databind")
 
   compileOnly(libs.microprofile.fault.tolerance.api)
+
+  compileOnly(project(":polaris-immutables"))
+  annotationProcessor(project(":polaris-immutables", configuration = "processor"))
 }
 
 val rootDir = rootProject.layout.projectDirectory
@@ -112,6 +116,10 @@ openApiGenerate {
       "IcebergErrorResponse" to "org.apache.iceberg.rest.responses.ErrorResponse",
       "OAuthError" to "org.apache.iceberg.rest.responses.ErrorResponse",
 
+      // S3 Signing
+      "S3SignRequest" to "org.apache.polaris.service.types.S3SignRequest",
+      "SignS3Request200Response" to "org.apache.polaris.service.types.S3SignResponse",
+
       // Custom types defined below
       "CommitViewRequest" to "org.apache.polaris.service.types.CommitViewRequest",
       "TokenType" to "org.apache.polaris.service.types.TokenType",

api/iceberg-service/src/main/java/org/apache/polaris/service/types/S3SignRequest.java

@@ -0,0 +1,61 @@
+/*
+ * 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.polaris.service.types;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import jakarta.annotation.Nullable;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.rest.RESTRequest;
+import org.apache.polaris.immutables.PolarisImmutable;
+import org.immutables.value.Value;
+
+/**
+ * Request for S3 signing requests.
+ *
+ * <p>Copy of {@link org.apache.iceberg.aws.s3.signer.S3SignRequest}, because the original does not
+ * have Jackson annotations.
+ */
+@PolarisImmutable
+@JsonDeserialize(as = ImmutableS3SignRequest.class)
+@JsonSerialize(as = ImmutableS3SignRequest.class)
+public interface S3SignRequest extends RESTRequest {
+
+  String region();
+
+  String method();
+
+  URI uri();
+
+  Map<String, List<String>> headers();
+
+  Map<String, String> properties();
+
+  @Value.Default
+  @Nullable
+  default String body() {
+    return null;
+  }
+
+  @Override
+  default void validate() {}
+}

api/iceberg-service/src/main/java/org/apache/polaris/service/types/S3SignResponse.java

@@ -0,0 +1,47 @@
+/*
+ * 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.polaris.service.types;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.rest.RESTResponse;
+import org.apache.polaris.immutables.PolarisImmutable;
+
+/**
+ * Response for S3 signing requests.
+ *
+ * <p>Copy of {@link org.apache.iceberg.aws.s3.signer.S3SignResponse}, because the original does not
+ * have Jackson annotations.
+ */
+@PolarisImmutable
+@JsonDeserialize(as = ImmutableS3SignResponse.class)
+@JsonSerialize(as = ImmutableS3SignResponse.class)
+public interface S3SignResponse extends RESTResponse {
+
+  URI uri();
+
+  Map<String, List<String>> headers();
+
+  @Override
+  default void validate() {}
+}

integration-tests/build.gradle.kts

@@ -33,6 +33,7 @@ dependencies {
   implementation(platform(libs.iceberg.bom))
   implementation("org.apache.iceberg:iceberg-api")
   implementation("org.apache.iceberg:iceberg-core")
+  implementation("org.apache.iceberg:iceberg-aws")
 
   implementation("org.apache.iceberg:iceberg-api:${libs.versions.iceberg.get()}:tests")
   implementation("org.apache.iceberg:iceberg-core:${libs.versions.iceberg.get()}:tests")

polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizableOperation.java

@@ -80,6 +80,7 @@
 import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_MANAGE_GRANTS_ON_SECURABLE;
 import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_READ_DATA;
 import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_READ_PROPERTIES;
+import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_REMOTE_SIGN;
 import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_WRITE_DATA;
 import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_WRITE_PROPERTIES;
 import static org.apache.polaris.core.entity.PolarisPrivilege.VIEW_CREATE;
@@ -212,7 +213,9 @@ public enum PolarisAuthorizableOperation {
   GET_APPLICABLE_POLICIES_ON_TABLE(TABLE_READ_PROPERTIES),
   ADD_POLICY_GRANT_TO_CATALOG_ROLE(POLICY_MANAGE_GRANTS_ON_SECURABLE),
   REVOKE_POLICY_GRANT_FROM_CATALOG_ROLE(
-      POLICY_MANAGE_GRANTS_ON_SECURABLE, CATALOG_ROLE_MANAGE_GRANTS_FOR_GRANTEE);
+      POLICY_MANAGE_GRANTS_ON_SECURABLE, CATALOG_ROLE_MANAGE_GRANTS_FOR_GRANTEE),
+  SIGN_S3_REQUEST(TABLE_REMOTE_SIGN),
+  ;
 
   private final EnumSet<PolarisPrivilege> privilegesOnTarget;
   private final EnumSet<PolarisPrivilege> privilegesOnSecondary;

polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizerImpl.java

@@ -92,6 +92,7 @@
 import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_MANAGE_GRANTS_ON_SECURABLE;
 import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_READ_DATA;
 import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_READ_PROPERTIES;
+import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_REMOTE_SIGN;
 import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_WRITE_DATA;
 import static org.apache.polaris.core.entity.PolarisPrivilege.TABLE_WRITE_PROPERTIES;
 import static org.apache.polaris.core.entity.PolarisPrivilege.VIEW_CREATE;
@@ -126,7 +127,7 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * Performs hierarchical resolution logic by matching the transively expanded set of grants to a
+ * Performs hierarchical resolution logic by matching the transitively expanded set of grants to a
  * calling principal against the cascading permissions over the parent hierarchy of a target
  * Securable.
  *
@@ -266,6 +267,10 @@ public class PolarisAuthorizerImpl implements PolarisAuthorizer {
     SUPER_PRIVILEGES.putAll(
         VIEW_FULL_METADATA,
         List.of(CATALOG_MANAGE_CONTENT, CATALOG_MANAGE_METADATA, VIEW_FULL_METADATA));
+    SUPER_PRIVILEGES.putAll(
+        TABLE_REMOTE_SIGN,
+        List.of(
+            CATALOG_MANAGE_CONTENT, CATALOG_MANAGE_METADATA, TABLE_CREATE, TABLE_FULL_METADATA));
 
     // Catalog privileges
     SUPER_PRIVILEGES.putAll(

polaris-core/src/main/java/org/apache/polaris/core/entity/PolarisPrivilege.java

@@ -155,6 +155,7 @@ public enum PolarisPrivilege {
       PolarisEntityType.POLICY,
       PolarisEntitySubType.NULL_SUBTYPE,
       PolarisEntityType.CATALOG_ROLE),
+  TABLE_REMOTE_SIGN(85, PolarisEntityType.NAMESPACE),
   ;
 
   /**

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogAdapter.java

@@ -18,6 +18,7 @@
  */
 package org.apache.polaris.service.catalog.iceberg;
 
+import static org.apache.polaris.service.catalog.AccessDelegationMode.REMOTE_SIGNING;
 import static org.apache.polaris.service.catalog.AccessDelegationMode.VENDED_CREDENTIALS;
 import static org.apache.polaris.service.catalog.validation.IcebergPropertiesValidation.validateIcebergProperties;
 
@@ -32,12 +33,14 @@
 import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.Response;
 import jakarta.ws.rs.core.SecurityContext;
+import jakarta.ws.rs.core.UriInfo;
 import java.util.EnumSet;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
 import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.aws.s3.signer.S3SignResponse;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.BadRequestException;
@@ -81,9 +84,11 @@
 import org.apache.polaris.service.context.catalog.CallContextCatalogFactory;
 import org.apache.polaris.service.http.IcebergHttpUtil;
 import org.apache.polaris.service.http.IfNoneMatch;
+import org.apache.polaris.service.storage.aws.signer.S3RequestSignerFactory;
 import org.apache.polaris.service.types.CommitTableRequest;
 import org.apache.polaris.service.types.CommitViewRequest;
 import org.apache.polaris.service.types.NotificationRequest;
+import org.apache.polaris.service.types.S3SignRequest;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -144,6 +149,8 @@ public class IcebergCatalogAdapter
   private final CatalogPrefixParser prefixParser;
   private final ReservedProperties reservedProperties;
   private final CatalogHandlerUtils catalogHandlerUtils;
+  private final S3RequestSignerFactory s3RequestSignerFactory;
+  private final UriInfo uriInfo;
 
   @Inject
   public IcebergCatalogAdapter(
@@ -157,7 +164,9 @@ public IcebergCatalogAdapter(
       PolarisAuthorizer polarisAuthorizer,
       CatalogPrefixParser prefixParser,
       ReservedProperties reservedProperties,
-      CatalogHandlerUtils catalogHandlerUtils) {
+      CatalogHandlerUtils catalogHandlerUtils,
+      S3RequestSignerFactory s3RequestSignerFactory,
+      UriInfo uriInfo) {
     this.realmContext = realmContext;
     this.callContext = callContext;
     this.catalogFactory = catalogFactory;
@@ -169,6 +178,8 @@ public IcebergCatalogAdapter(
     this.prefixParser = prefixParser;
     this.reservedProperties = reservedProperties;
     this.catalogHandlerUtils = catalogHandlerUtils;
+    this.s3RequestSignerFactory = s3RequestSignerFactory;
+    this.uriInfo = uriInfo;
   }
 
   /**
@@ -205,7 +216,9 @@ IcebergCatalogHandler newHandlerWrapper(SecurityContext securityContext, String
         catalogName,
         polarisAuthorizer,
         reservedProperties,
-        catalogHandlerUtils);
+        catalogHandlerUtils,
+        s3RequestSignerFactory,
+        uriInfo);
   }
 
   @Override
@@ -323,11 +336,13 @@ public Response updateProperties(
         catalog -> Response.ok(catalog.updateNamespaceProperties(ns, revisedRequest)).build());
   }
 
-  private EnumSet<AccessDelegationMode> parseAccessDelegationModes(String accessDelegationMode) {
+  private static Set<AccessDelegationMode> parseAccessDelegationModes(String accessDelegationMode) {
     EnumSet<AccessDelegationMode> delegationModes =
         AccessDelegationMode.fromProtocolValuesList(accessDelegationMode);
     Preconditions.checkArgument(
-        delegationModes.isEmpty() || delegationModes.contains(VENDED_CREDENTIALS),
+        delegationModes.isEmpty()
+            || delegationModes.contains(VENDED_CREDENTIALS)
+            || delegationModes.contains(REMOTE_SIGNING),
         "Unsupported access delegation mode: %s",
         accessDelegationMode);
     return delegationModes;
@@ -342,8 +357,7 @@ public Response createTable(
       RealmContext realmContext,
       SecurityContext securityContext) {
     validateIcebergProperties(callContext, createTableRequest.properties());
-    EnumSet<AccessDelegationMode> delegationModes =
-        parseAccessDelegationModes(accessDelegationMode);
+    Set<AccessDelegationMode> delegationModes = parseAccessDelegationModes(accessDelegationMode);
     Namespace ns = decodeNamespace(namespace);
     return withCatalog(
         securityContext,
@@ -354,7 +368,8 @@ public Response createTable(
               return Response.ok(catalog.createTableStaged(ns, createTableRequest)).build();
             } else {
               return Response.ok(
-                      catalog.createTableStagedWithWriteDelegation(ns, createTableRequest))
+                      catalog.createTableStagedWithWriteDelegation(
+                          prefix, ns, createTableRequest, delegationModes))
                   .build();
             }
           } else if (delegationModes.isEmpty()) {
@@ -364,7 +379,8 @@ public Response createTable(
                 .build();
           } else {
             LoadTableResponse response =
-                catalog.createTableDirectWithWriteDelegation(ns, createTableRequest);
+                catalog.createTableDirectWithWriteDelegation(
+                    prefix, ns, createTableRequest, delegationModes);
             return tryInsertETagHeader(
                     Response.ok(response), response, namespace, createTableRequest.name())
                 .build();
@@ -397,8 +413,7 @@ public Response loadTable(
       String snapshots,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    EnumSet<AccessDelegationMode> delegationModes =
-        parseAccessDelegationModes(accessDelegationMode);
+    Set<AccessDelegationMode> delegationModes = parseAccessDelegationModes(accessDelegationMode);
     Namespace ns = decodeNamespace(namespace);
     TableIdentifier tableIdentifier = TableIdentifier.of(ns, RESTUtil.decodeString(table));
 
@@ -422,7 +437,8 @@ public Response loadTable(
           } else {
             response =
                 catalog
-                    .loadTableWithAccessDelegationIfStale(tableIdentifier, ifNoneMatch, snapshots)
+                    .loadTableWithAccessDelegationIfStale(
+                        prefix, tableIdentifier, ifNoneMatch, delegationModes)
                     .orElseThrow(() -> new WebApplicationException(Response.Status.NOT_MODIFIED));
           }
 
@@ -588,8 +604,7 @@ public Response loadCredentials(
         securityContext,
         prefix,
         catalog -> {
-          LoadTableResponse loadTableResponse =
-              catalog.loadTableWithAccessDelegation(tableIdentifier, "all");
+          LoadTableResponse loadTableResponse = catalog.loadTable(tableIdentifier, "all");
           return Response.ok(
                   ImmutableLoadCredentialsResponse.builder()
                       .credentials(loadTableResponse.credentials())
@@ -797,4 +812,24 @@ public Response getConfig(
                 .build())
         .build();
   }
+
+  @Override
+  public Response signS3Request(
+      String prefix,
+      String namespace,
+      String table,
+      S3SignRequest s3SignRequest,
+      RealmContext realmContext,
+      SecurityContext securityContext) {
+    Namespace ns = decodeNamespace(namespace);
+    TableIdentifier tableIdentifier = TableIdentifier.of(ns, RESTUtil.decodeString(table));
+    return withCatalog(
+        securityContext,
+        prefix,
+        catalog -> {
+          // TODO Cache-Control header
+          S3SignResponse response = catalog.signS3Request(s3SignRequest, tableIdentifier);
+          return Response.status(Response.Status.OK).entity(response).build();
+        });
+  }
 }