From ad0f0d03e7e2a716b9dccd50321cc00f620c5431 Mon Sep 17 00:00:00 2001 From: Alexandre Dutra Date: Fri, 1 Aug 2025 15:15:25 +0200 Subject: [PATCH] Add support for S3 request signing Fixes #32. --- CHANGELOG.md | 8 + LICENSE | 1 + README.md | 1 + api/README.md | 2 + api/s3-sign-service/build.gradle.kts | 114 +++++++++ .../s3/sign/model/PolarisS3SignRequest.java | 53 +++++ .../s3/sign/model/PolarisS3SignResponse.java | 36 +++ bom/build.gradle.kts | 1 + build.gradle.kts | 1 + gradle/projects.main.properties | 1 + ...PolarisS3RemoteSigningIntegrationTest.java | 151 ++++++++++++ .../auth/PolarisAuthorizableOperation.java | 6 +- .../core/auth/PolarisAuthorizerImpl.java | 7 +- .../core/config/FeatureConfiguration.java | 10 + .../polaris/core/entity/PolarisPrivilege.java | 1 + .../polaris/core/rest/PolarisEndpoints.java | 19 ++ .../core/rest/PolarisResourcePaths.java | 15 ++ .../polaris/core/storage/AccessConfig.java | 3 + .../core/storage/StorageAccessProperty.java | 18 ++ .../aws/AwsCredentialsStorageIntegration.java | 23 +- runtime/service/build.gradle.kts | 1 + .../service/it/S3RemoteSigningMinIOIT.java | 93 ++++++++ .../catalog/common/CatalogHandler.java | 7 + .../catalog/iceberg/IcebergCatalog.java | 78 ++++++- .../iceberg/IcebergCatalogAdapter.java | 36 +-- .../iceberg/IcebergCatalogHandler.java | 83 ++++--- .../iceberg/SupportsCredentialDelegation.java | 2 +- .../iceberg/SupportsRemoteSigning.java | 28 +++ .../catalog/io/DefaultFileIOFactory.java | 39 +++- .../service/catalog/io/FileIOUtil.java | 14 +- .../io/WasbTranslatingFileIOFactory.java | 9 +- .../PolarisCallContextCatalogFactory.java | 19 +- ...PolarisStorageIntegrationProviderImpl.java | 2 +- .../service/storage/StorageConfiguration.java | 23 +- .../sign/S3RemoteSigningCatalogAdapter.java | 100 ++++++++ .../sign/S3RemoteSigningCatalogHandler.java | 119 ++++++++++ .../storage/s3/sign/S3RequestSigner.java | 30 +++ .../storage/s3/sign/S3RequestSignerImpl.java | 84 +++++++ .../service/admin/PolarisAuthzTestBase.java | 27 ++- .../admin/PolarisS3InteroperabilityTest.java | 2 +- .../catalog/AbstractIcebergCatalogTest.java | 22 +- .../AbstractIcebergCatalogViewTest.java | 15 +- ...bstractPolarisGenericTableCatalogTest.java | 14 +- .../catalog/AbstractPolicyCatalogTest.java | 14 +- .../IcebergCatalogHandlerAuthzTest.java | 100 ++++++-- .../service/catalog/io/FileIOFactoryTest.java | 21 +- .../S3RemoteSigningMinIOIntegrationTest.java | 93 ++++++++ .../storage/StorageConfigurationTest.java | 4 +- ...3RemoteSigningCatalogHandlerAuthzTest.java | 120 ++++++++++ .../s3/sign/S3RequestSignerImplTest.java | 221 ++++++++++++++++++ .../apache/polaris/service/TestServices.java | 25 +- .../catalog/io/MeasuredFileIOFactory.java | 9 +- .../in-dev/unreleased/access-control.md | 26 ++- spec/README.md | 42 +++- spec/s3-sign/iceberg-s3-signer-open-api.yaml | 154 ++++++++++++ spec/s3-sign/polaris-s3-sign-service.yaml | 108 +++++++++ 56 files changed, 2113 insertions(+), 142 deletions(-) create mode 100644 api/s3-sign-service/build.gradle.kts create mode 100644 api/s3-sign-service/src/main/java/org/apache/polaris/service/s3/sign/model/PolarisS3SignRequest.java create mode 100644 api/s3-sign-service/src/main/java/org/apache/polaris/service/s3/sign/model/PolarisS3SignResponse.java create mode 100644 integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisS3RemoteSigningIntegrationTest.java create mode 100644 runtime/service/src/intTest/java/org/apache/polaris/service/it/S3RemoteSigningMinIOIT.java create mode 100644 runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/SupportsRemoteSigning.java create mode 100644 runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogAdapter.java create mode 100644 runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogHandler.java create mode 100644 runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RequestSigner.java create mode 100644 runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RequestSignerImpl.java create mode 100644 runtime/service/src/test/java/org/apache/polaris/service/it/S3RemoteSigningMinIOIntegrationTest.java create mode 100644 runtime/service/src/test/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogHandlerAuthzTest.java create mode 100644 runtime/service/src/test/java/org/apache/polaris/service/storage/s3/sign/S3RequestSignerImplTest.java create mode 100644 spec/s3-sign/iceberg-s3-signer-open-api.yaml create mode 100644 spec/s3-sign/polaris-s3-sign-service.yaml diff --git a/CHANGELOG.md b/CHANGELOG.md index fdbe0c1b0f..e2e8cba092 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,14 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti ### Highlights +- Support for S3 remote request signing has been added, allowing Polaris to work with S3-compatible object storage systems. + *Remote signing is currently experimental and not enabled by default*. In particular, RBAC checks are currently not + production-ready. One new table privilege was introduced: `TABLE_REMOTE_SIGN`. To enable remote signing: + 1. Set the system-wide property `REMOTE_SIGNING_ENABLED` or the catalog-level `polaris.request-signing.enabled` + property to `true`. + 2. Grant the `TABLE_REMOTE_SIGN` privilege to a catalog role. The role must also be granted the `TABLE_READ_DATA` + and `TABLE_WRITE_DATA` privileges. + ### Upgrade notes ### Breaking changes diff --git a/LICENSE b/LICENSE index 54fe69af67..5c8286cbb3 100644 --- a/LICENSE +++ b/LICENSE @@ -217,6 +217,7 @@ This product includes code from Apache Iceberg. * spec/iceberg-rest-catalog-open-api.yaml * spec/polaris-catalog-apis/oauth-tokens-api.yaml +* spec/s3-sign/iceberg-s3-signer-open-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 * runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/CatalogHandlerUtils.java diff --git a/README.md b/README.md index 3446ee64ff..e31454f254 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,7 @@ Apache Polaris is organized into the following modules: - `polaris-api-management-model` - The Polaris management model - `polaris-api-management-service` - The Polaris management service - `polaris-api-iceberg-service` - The Iceberg REST service + - `polaris-api-s3-sign-service` - The Iceberg REST service for S3 remote signing - Runtime modules: - `polaris-runtime-service` - The runtime components of the Polaris server - `polaris-runtime-defaults` - The runtime configuration defaults diff --git a/api/README.md b/api/README.md index 7c7fb61fc0..d8093d99be 100644 --- a/api/README.md +++ b/api/README.md @@ -33,6 +33,8 @@ This directory contains the API modules for Apache Polaris. Iceberg REST API. - [`polaris-api-catalog-service`](polaris-catalog-service): contains the service classes for the Polaris native Catalog REST API. +- [`polaris-api-s3-sign-service`](s3-sign-service): contains the model and service classes + for the S3 remote signing REST API. The classes in these modules are generated from the OpenAPI specification files in the [`spec`](../spec) directory. \ No newline at end of file diff --git a/api/s3-sign-service/build.gradle.kts b/api/s3-sign-service/build.gradle.kts new file mode 100644 index 0000000000..36076b85f8 --- /dev/null +++ b/api/s3-sign-service/build.gradle.kts @@ -0,0 +1,114 @@ +/* + * 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. + */ + +import org.openapitools.generator.gradle.plugin.tasks.GenerateTask + +plugins { + alias(libs.plugins.openapi.generator) + id("polaris-client") + alias(libs.plugins.jandex) +} + +dependencies { + implementation(project(":polaris-core")) + + 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) + implementation(libs.jakarta.validation.api) + implementation(libs.swagger.annotations) + + implementation(libs.jakarta.servlet.api) + implementation(libs.jakarta.ws.rs.api) + + implementation(platform(libs.micrometer.bom)) + implementation("io.micrometer:micrometer-core") + + implementation(platform(libs.jackson.bom)) + implementation("com.fasterxml.jackson.core:jackson-annotations") + implementation("com.fasterxml.jackson.core:jackson-core") + 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 +val specsDir = rootDir.dir("spec") +val templatesDir = rootDir.dir("server-templates") +// Use a different directory than 'generated/', because OpenAPI generator's `GenerateTask` adds the +// whole directory to its task output, but 'generated/' is not exclusive to that task and in turn +// breaks Gradle's caching. +val generatedDir = project.layout.buildDirectory.dir("generated-openapi") +val generatedOpenApiSrcDir = project.layout.buildDirectory.dir("generated-openapi/src/main/java") + +openApiGenerate { + // The OpenAPI generator does NOT resolve relative paths correctly against the Gradle project + // directory + inputSpec = provider { specsDir.file("s3-sign/polaris-s3-sign-service.yaml").asFile.absolutePath } + generatorName = "jaxrs-resteasy" + outputDir = provider { generatedDir.get().asFile.absolutePath } + apiPackage = "org.apache.polaris.service.s3.sign.api" + ignoreFileOverride.set(provider { rootDir.file(".openapi-generator-ignore").asFile.absolutePath }) + templateDir.set(provider { templatesDir.asFile.absolutePath }) + removeOperationIdPrefix.set(true) + globalProperties.put("apis", "S3SignerApi") + globalProperties.put("models", "false") + globalProperties.put("apiDocs", "false") + globalProperties.put("modelTests", "false") + configOptions.put("resourceName", "catalog") + configOptions.put("useTags", "true") + configOptions.put("useBeanValidation", "false") + configOptions.put("sourceFolder", "src/main/java") + configOptions.put("useJakartaEe", "true") + configOptions.put("hideGenerationTimestamp", "true") + additionalProperties.put("apiNamePrefix", "IcebergRest") + additionalProperties.put("apiNameSuffix", "") + additionalProperties.put("metricsPrefix", "polaris") + serverVariables.put("basePath", "api/s3-sign") + modelNameMappings = mapOf("S3SignRequest" to "PolarisS3SignRequest") + typeMappings = + mapOf("S3SignRequest" to "org.apache.polaris.service.s3.sign.model.PolarisS3SignRequest") + importMappings = + mapOf( + "IcebergErrorResponse" to "org.apache.iceberg.rest.responses.ErrorResponse", + "PolarisS3SignRequest" to "org.apache.polaris.service.s3.sign.model.PolarisS3SignRequest", + "SignS3Request200Response" to "org.apache.polaris.service.s3.sign.model.PolarisS3SignResponse", + ) +} + +listOf("sourcesJar", "compileJava", "processResources").forEach { task -> + tasks.named(task) { dependsOn("openApiGenerate") } +} + +sourceSets { main { java { srcDir(generatedOpenApiSrcDir) } } } + +tasks.named("openApiGenerate") { + inputs.dir(templatesDir) + inputs.dir(specsDir) + actions.addFirst { delete { delete(generatedDir) } } +} + +tasks.named("javadoc") { dependsOn("jandex") } diff --git a/api/s3-sign-service/src/main/java/org/apache/polaris/service/s3/sign/model/PolarisS3SignRequest.java b/api/s3-sign-service/src/main/java/org/apache/polaris/service/s3/sign/model/PolarisS3SignRequest.java new file mode 100644 index 0000000000..efc9d63746 --- /dev/null +++ b/api/s3-sign-service/src/main/java/org/apache/polaris/service/s3/sign/model/PolarisS3SignRequest.java @@ -0,0 +1,53 @@ +/* + * 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.s3.sign.model; + +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import jakarta.annotation.Nullable; +import org.apache.iceberg.aws.s3.signer.S3SignRequest; +import org.apache.polaris.immutables.PolarisImmutable; +import org.immutables.value.Value; + +/** + * Request for S3 signing. + * + *

Copy of {@link S3SignRequest}, because the original does not have Jackson annotations. + */ +@PolarisImmutable +@JsonDeserialize(as = ImmutablePolarisS3SignRequest.class) +@JsonSerialize(as = ImmutablePolarisS3SignRequest.class) +@SuppressWarnings("immutables:subtype") +public interface PolarisS3SignRequest extends S3SignRequest { + + @Value.Default + @Nullable // Replace javax.annotation.Nullable from S3SignRequest with jakarta.annotation.Nullable + @Override + default String body() { + return null; + } + + default boolean write() { + return method().equalsIgnoreCase("PUT") + || method().equalsIgnoreCase("POST") + || method().equalsIgnoreCase("DELETE") + || method().equalsIgnoreCase("PATCH"); + } +} diff --git a/api/s3-sign-service/src/main/java/org/apache/polaris/service/s3/sign/model/PolarisS3SignResponse.java b/api/s3-sign-service/src/main/java/org/apache/polaris/service/s3/sign/model/PolarisS3SignResponse.java new file mode 100644 index 0000000000..9f4cb778cf --- /dev/null +++ b/api/s3-sign-service/src/main/java/org/apache/polaris/service/s3/sign/model/PolarisS3SignResponse.java @@ -0,0 +1,36 @@ +/* + * 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.s3.sign.model; + +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import org.apache.iceberg.aws.s3.signer.S3SignResponse; +import org.apache.polaris.immutables.PolarisImmutable; + +/** + * Response for S3 signing requests. + * + *

Copy of {@link S3SignResponse}, because the original does not have Jackson annotations. + */ +@PolarisImmutable +@JsonDeserialize(as = ImmutablePolarisS3SignResponse.class) +@JsonSerialize(as = ImmutablePolarisS3SignResponse.class) +@SuppressWarnings("immutables:subtype") +public interface PolarisS3SignResponse extends S3SignResponse {} diff --git a/bom/build.gradle.kts b/bom/build.gradle.kts index 0b37372c2f..18e7a2d5cc 100644 --- a/bom/build.gradle.kts +++ b/bom/build.gradle.kts @@ -28,6 +28,7 @@ dependencies { api(project(":polaris-api-iceberg-service")) api(project(":polaris-api-management-model")) api(project(":polaris-api-management-service")) + api(project(":polaris-api-s3-sign-service")) api(project(":polaris-container-spec-helper")) api(project(":polaris-minio-testcontainer")) diff --git a/build.gradle.kts b/build.gradle.kts index 4d52ad6dac..30f35b9b62 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -152,6 +152,7 @@ tasks.register("regeneratePythonClient") { dependsOn(":polaris-api-management-service:processResources") dependsOn(":polaris-api-catalog-service:processResources") dependsOn(":polaris-api-management-model:processResources") + dependsOn(":polaris-api-s3-sign-service:processResources") } // Pass environment variables: diff --git a/gradle/projects.main.properties b/gradle/projects.main.properties index 27fdae3556..6fe429a5b4 100644 --- a/gradle/projects.main.properties +++ b/gradle/projects.main.properties @@ -24,6 +24,7 @@ polaris-api-iceberg-service=api/iceberg-service polaris-api-management-model=api/management-model polaris-api-management-service=api/management-service polaris-api-catalog-service=api/polaris-catalog-service +polaris-api-s3-sign-service=api/s3-sign-service polaris-runtime-defaults=runtime/defaults polaris-runtime-service=runtime/service polaris-server=runtime/server diff --git a/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisS3RemoteSigningIntegrationTest.java b/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisS3RemoteSigningIntegrationTest.java new file mode 100644 index 0000000000..8a1e7506c2 --- /dev/null +++ b/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisS3RemoteSigningIntegrationTest.java @@ -0,0 +1,151 @@ +/* + * 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.it.test; + +import static org.apache.polaris.core.storage.StorageAccessProperty.AWS_ENDPOINT; +import static org.apache.polaris.core.storage.StorageAccessProperty.AWS_PATH_STYLE_ACCESS; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.google.common.collect.ImmutableMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableMetadataParser; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.ForbiddenException; +import org.apache.iceberg.io.ResolvingFileIO; +import org.apache.iceberg.rest.RESTCatalog; +import org.apache.polaris.core.admin.model.AwsStorageConfigInfo; +import org.apache.polaris.core.admin.model.Catalog; +import org.apache.polaris.core.admin.model.StorageConfigInfo; +import org.apache.polaris.core.config.FeatureConfiguration; +import org.apache.polaris.service.it.env.CatalogConfig; +import org.apache.polaris.service.it.env.RestCatalogConfig; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +/** Integration tests for S3 remote signing. */ +@CatalogConfig(properties = {"header.X-Iceberg-Access-Delegation", "remote-signing"}) +@RestCatalogConfig({ + // The default client file IO implementation is InMemoryFileIO, + // which does not support remote signing. + org.apache.iceberg.CatalogProperties.FILE_IO_IMPL, + "org.apache.iceberg.io.ResolvingFileIO", +}) +public abstract class PolarisS3RemoteSigningIntegrationTest + extends PolarisRestCatalogIntegrationBase { + + @Override + protected StorageConfigInfo getStorageConfigInfo() { + return AwsStorageConfigInfo.builder() + .setRoleArn(roleArn()) + .setStorageType(StorageConfigInfo.StorageTypeEnum.S3) + .setPathStyleAccess(pathStyleAccess()) + .setAllowedLocations(allowedLocations()) + .setEndpoint(endpoint().orElse(null)) + .build(); + } + + @Override + protected ImmutableMap.Builder clientFileIOProperties() { + ImmutableMap.Builder builder = + super.clientFileIOProperties() + .put(AWS_PATH_STYLE_ACCESS.getPropertyName(), String.valueOf(pathStyleAccess())); + endpoint().ifPresent(endpoint -> builder.put(AWS_ENDPOINT.getPropertyName(), endpoint)); + return builder; + } + + protected String roleArn() { + return "arn:aws:iam::123456789012:role/my-role"; + } + + protected boolean pathStyleAccess() { + return true; + } + + protected Optional endpoint() { + return Optional.empty(); + } + + protected abstract List allowedLocations(); + + @CatalogConfig(properties = {"polaris.config.remote-signing.enabled", "false"}) + @Test + public void testInternalCatalogRemoteSigningDisabled() { + @SuppressWarnings("resource") + RESTCatalog catalog = catalog(); + Namespace ns1 = Namespace.of("ns1"); + catalog.createNamespace(ns1); + TableIdentifier tableIdentifier = TableIdentifier.of(ns1, "my_table"); + assertThatThrownBy(() -> catalog.createTable(tableIdentifier, SCHEMA)) + .isInstanceOf(ForbiddenException.class) + .hasMessageContaining("Remote signing is not enabled for this catalog") + .hasMessageContaining(FeatureConfiguration.REMOTE_SIGNING_ENABLED.key()) + .hasMessageContaining(FeatureConfiguration.REMOTE_SIGNING_ENABLED.catalogConfig()); + } + + @CatalogConfig(Catalog.TypeEnum.EXTERNAL) + @Test + public void testExternalCatalogRemoteSigningDisabled() { + @SuppressWarnings("resource") + RESTCatalog catalog = catalog(); + Namespace ns1 = Namespace.of("ns1"); + catalog.createNamespace(ns1); + TableMetadata tableMetadata = + TableMetadata.newTableMetadata( + SCHEMA, + PartitionSpec.unpartitioned(), + externalCatalogBaseLocation() + "/ns1/my_table", + Map.of()); + try (ResolvingFileIO resolvingFileIO = initializeClientFileIO(new ResolvingFileIO())) { + String fileLocation = + externalCatalogBaseLocation() + "/ns1/my_table/metadata/v1.metadata.json"; + TableMetadataParser.write(tableMetadata, resolvingFileIO.newOutputFile(fileLocation)); + catalog.registerTable(TableIdentifier.of(ns1, "my_table"), fileLocation); + try { + assertThatThrownBy(() -> catalog.loadTable(TableIdentifier.of(ns1, "my_table"))) + .isInstanceOf(ForbiddenException.class) + .hasMessageContaining("Remote signing is not enabled for external catalogs"); + } finally { + resolvingFileIO.deleteFile(fileLocation); + } + } + } + + @Test + @Override + @Disabled("It's not possible to request an access delegation mode when registering a table.") + public void testRegisterTable() { + // FIXME this test should work if Polaris could send the right AccessConfig even if no + // delegation mode was requested when registering the table. + } + + @Test + @Override + @Disabled("This test is explicitly for vended credentials") + public void testLoadTableWithAccessDelegationForExternalCatalogWithConfigDisabled() {} + + @Test + @Override + @Disabled("This test is explicitly for vended credentials") + public void testLoadTableWithoutAccessDelegationForExternalCatalogWithConfigDisabled() {} +} diff --git a/polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizableOperation.java b/polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizableOperation.java index 2013b4f28e..35d615cfd9 100644 --- a/polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizableOperation.java +++ b/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,10 @@ 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_READ_REQUEST(EnumSet.of(TABLE_REMOTE_SIGN, TABLE_READ_DATA)), + SIGN_S3_WRITE_REQUEST(EnumSet.of(TABLE_REMOTE_SIGN, TABLE_WRITE_DATA)), + ; private final EnumSet privilegesOnTarget; private final EnumSet privilegesOnSecondary; diff --git a/polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizerImpl.java b/polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizerImpl.java index 72be7f88cb..878d9f65aa 100644 --- a/polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizerImpl.java +++ b/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( diff --git a/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java b/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java index fe5d6bc6a8..9541e68de3 100644 --- a/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java +++ b/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java @@ -348,4 +348,14 @@ public static void enforceFeatureEnabledOrThrow( + "it is still possible to enforce the uniqueness of table locations within a catalog.") .defaultValue(false) .buildFeatureConfiguration(); + + public static final FeatureConfiguration REMOTE_SIGNING_ENABLED = + PolarisConfiguration.builder() + .key("REMOTE_SIGNING_ENABLED") + .catalogConfig("polaris.config.remote-signing.enabled") + .description( + "If true, the remote signing endpoints are enabled either globally, or for a specific catalog. " + + "This feature is currently experimental and may change in future releases.") + .defaultValue(false) + .buildFeatureConfiguration(); } diff --git a/polaris-core/src/main/java/org/apache/polaris/core/entity/PolarisPrivilege.java b/polaris-core/src/main/java/org/apache/polaris/core/entity/PolarisPrivilege.java index 88cf6083bf..be1cf3b3d7 100644 --- a/polaris-core/src/main/java/org/apache/polaris/core/entity/PolarisPrivilege.java +++ b/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.TABLE_LIKE, PolarisEntitySubType.ICEBERG_TABLE), ; /** diff --git a/polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisEndpoints.java b/polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisEndpoints.java index 2bae38ff7e..5b8f85e0d0 100644 --- a/polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisEndpoints.java +++ b/polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisEndpoints.java @@ -23,6 +23,7 @@ import org.apache.iceberg.rest.Endpoint; import org.apache.polaris.core.config.FeatureConfiguration; import org.apache.polaris.core.config.RealmConfig; +import org.apache.polaris.core.entity.CatalogEntity; public class PolarisEndpoints { // Generic table endpoints @@ -73,6 +74,12 @@ public class PolarisEndpoints { .add(V1_GET_APPLICABLE_POLICIES) .build(); + public static final Endpoint V1_S3_REMOTE_SIGNING = + Endpoint.create("POST", PolarisResourcePaths.V1_S3_REMOTE_SIGNING); + + public static final Set REMOTE_SIGNING_ENDPOINTS = + ImmutableSet.builder().add(V1_S3_REMOTE_SIGNING).build(); + /** * Get the generic table endpoints. Returns GENERIC_TABLE_ENDPOINTS if ENABLE_GENERIC_TABLES is * set to true, otherwise, returns an empty set. @@ -92,4 +99,16 @@ public static Set getSupportedPolicyEndpoints(RealmConfig realmConfig) boolean policyStoreEnabled = realmConfig.getConfig(FeatureConfiguration.ENABLE_POLICY_STORE); return policyStoreEnabled ? POLICY_STORE_ENDPOINTS : ImmutableSet.of(); } + + /** + * Get the remote signing endpoints. Returns {@link #REMOTE_SIGNING_ENDPOINTS} if {@link + * FeatureConfiguration#REMOTE_SIGNING_ENABLED} is set globally to true or if the catalog enables + * remote signing; otherwise, returns an empty set. + */ + public static Set getSupportedRemoteSigningEndpoints( + RealmConfig realmConfig, CatalogEntity catalogEntity) { + boolean remoteSigningEnabled = + realmConfig.getConfig(FeatureConfiguration.REMOTE_SIGNING_ENABLED, catalogEntity); + return remoteSigningEnabled ? REMOTE_SIGNING_ENDPOINTS : ImmutableSet.of(); + } } diff --git a/polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisResourcePaths.java b/polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisResourcePaths.java index 8a30d79624..42e73dbb45 100644 --- a/polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisResourcePaths.java +++ b/polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisResourcePaths.java @@ -42,6 +42,10 @@ public class PolarisResourcePaths { "/polaris/v1/{prefix}/namespaces/{namespace}/policies/{policy-name}/mappings"; public static final String V1_APPLICABLE_POLICIES = "/polaris/v1/{prefix}/applicable-policies"; + // S3 Remote Signing endpoint + public static final String V1_S3_REMOTE_SIGNING = + "/s3-sign/v1/{prefix}/namespaces/{namespace}/tables/{table}"; + private final String prefix; public PolarisResourcePaths(String prefix) { @@ -67,4 +71,15 @@ public String genericTable(TableIdentifier ident) { "generic-tables", RESTUtil.encodeString(ident.name())); } + + public String s3RemoteSigning(TableIdentifier ident) { + return SLASH.join( + "s3-sign", + "v1", + prefix, + "namespaces", + RESTUtil.encodeNamespace(ident.namespace()), + "tables", + RESTUtil.encodeString(ident.name())); + } } diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/AccessConfig.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/AccessConfig.java index e15fd2e916..b725afaeb5 100644 --- a/polaris-core/src/main/java/org/apache/polaris/core/storage/AccessConfig.java +++ b/polaris-core/src/main/java/org/apache/polaris/core/storage/AccessConfig.java @@ -26,6 +26,9 @@ @PolarisImmutable public interface AccessConfig { + + AccessConfig EMPTY = AccessConfig.builder().build(); + Map credentials(); Map extraProperties(); diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/StorageAccessProperty.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/StorageAccessProperty.java index 33526d2e29..06314dca1e 100644 --- a/polaris-core/src/main/java/org/apache/polaris/core/storage/StorageAccessProperty.java +++ b/polaris-core/src/main/java/org/apache/polaris/core/storage/StorageAccessProperty.java @@ -18,6 +18,9 @@ */ package org.apache.polaris.core.storage; +import org.apache.iceberg.aws.s3.S3FileIOProperties; +import org.apache.iceberg.aws.s3.signer.S3V4RestSignerClient; + /** * A subset of Iceberg catalog properties recognized by Polaris. * @@ -39,6 +42,21 @@ public enum StorageAccessProperty { Boolean.class, "s3.path-style-access", "whether to use S3 path style access", false), CLIENT_REGION( String.class, "client.region", "region to configure client for making requests to AWS"), + AWS_REMOTE_SIGNING_ENABLED( + Boolean.class, + S3FileIOProperties.REMOTE_SIGNING_ENABLED, + "whether to enable remote signing for S3 requests", + false), + AWS_REMOTE_SIGNER_URI( + String.class, + S3V4RestSignerClient.S3_SIGNER_URI, + "the base URI for the remote signer service, used for signing S3 requests", + false), + AWS_REMOTE_SIGNER_ENDPOINT( + String.class, + S3V4RestSignerClient.S3_SIGNER_ENDPOINT, + "the endpoint for the remote signer service, used for signing S3 requests", + false), GCS_ACCESS_TOKEN(String.class, "gcs.oauth2.token", "the gcs scoped access token"), GCS_ACCESS_TOKEN_EXPIRES_AT( diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsCredentialsStorageIntegration.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsCredentialsStorageIntegration.java index 616fb1f4d5..9b35251258 100644 --- a/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsCredentialsStorageIntegration.java +++ b/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsCredentialsStorageIntegration.java @@ -116,6 +116,27 @@ public AccessConfig getSubscopedCreds( String.valueOf(i.toEpochMilli())); }); + addCommonProperties(region, accessConfig, storageConfig); + + return accessConfig.build(); + } + + public AccessConfig getRemoteSigningAccessConfig(URI signerUri, String signerEndpoint) { + + AwsStorageConfigurationInfo storageConfig = config(); + AccessConfig.Builder accessConfig = AccessConfig.builder(); + + accessConfig.put(StorageAccessProperty.AWS_REMOTE_SIGNING_ENABLED, Boolean.TRUE.toString()); + accessConfig.put(StorageAccessProperty.AWS_REMOTE_SIGNER_URI, signerUri.toString()); + accessConfig.put(StorageAccessProperty.AWS_REMOTE_SIGNER_ENDPOINT, signerEndpoint); + + addCommonProperties(storageConfig.getRegion(), accessConfig, storageConfig); + + return accessConfig.build(); + } + + private static void addCommonProperties( + String region, AccessConfig.Builder accessConfig, AwsStorageConfigurationInfo storageConfig) { if (region != null) { accessConfig.put(StorageAccessProperty.CLIENT_REGION, region); } @@ -139,8 +160,6 @@ public AccessConfig getSubscopedCreds( String.format( "AWS region must be set when using partition %s", storageConfig.getAwsPartition())); } - - return accessConfig.build(); } /** diff --git a/runtime/service/build.gradle.kts b/runtime/service/build.gradle.kts index 87e95c5300..f573fba55f 100644 --- a/runtime/service/build.gradle.kts +++ b/runtime/service/build.gradle.kts @@ -30,6 +30,7 @@ dependencies { implementation(project(":polaris-api-management-service")) implementation(project(":polaris-api-iceberg-service")) implementation(project(":polaris-api-catalog-service")) + implementation(project(":polaris-api-s3-sign-service")) runtimeOnly(project(":polaris-relational-jdbc")) diff --git a/runtime/service/src/intTest/java/org/apache/polaris/service/it/S3RemoteSigningMinIOIT.java b/runtime/service/src/intTest/java/org/apache/polaris/service/it/S3RemoteSigningMinIOIT.java new file mode 100644 index 0000000000..a64f6bb4de --- /dev/null +++ b/runtime/service/src/intTest/java/org/apache/polaris/service/it/S3RemoteSigningMinIOIT.java @@ -0,0 +1,93 @@ +/* + * 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.it; + +import com.google.common.collect.ImmutableMap; +import io.quarkus.test.junit.QuarkusIntegrationTest; +import io.quarkus.test.junit.QuarkusTestProfile; +import io.quarkus.test.junit.TestProfile; +import java.net.URI; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.apache.polaris.core.storage.StorageAccessProperty; +import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension; +import org.apache.polaris.service.it.test.PolarisS3RemoteSigningIntegrationTest; +import org.apache.polaris.test.minio.Minio; +import org.apache.polaris.test.minio.MinioAccess; +import org.apache.polaris.test.minio.MinioExtension; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.extension.ExtendWith; + +@QuarkusIntegrationTest +@TestProfile(S3RemoteSigningMinIOIT.Profile.class) +@ExtendWith(MinioExtension.class) +@ExtendWith(PolarisIntegrationTestExtension.class) +public class S3RemoteSigningMinIOIT extends PolarisS3RemoteSigningIntegrationTest { + + private static final String BUCKET_URI_PREFIX = "/minio-test"; + private static final String MINIO_ACCESS_KEY = "test-ak-123"; + private static final String MINIO_SECRET_KEY = "test-sk-123"; + + public static class Profile implements QuarkusTestProfile { + + @Override + public Map getConfigOverrides() { + return ImmutableMap.builder() + // Polaris will use these keys for remote signing + .put("polaris.storage.aws.access-key", MINIO_ACCESS_KEY) + .put("polaris.storage.aws.secret-key", MINIO_SECRET_KEY) + .put("polaris.features.\"REMOTE_SIGNING_ENABLED\"", "true") + .put("polaris.features.\"SUPPORTED_CATALOG_STORAGE_TYPES\"", "[\"S3\"]") + // Required for MinIO because Polaris needs subscoped credentials + // in order to get the right endpoint URL to use when writing metadata files. + .put("polaris.features.\"SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION\"", "false") + .build(); + } + } + + private static URI storageBase; + private static String endpoint; + + @BeforeAll + static void setup( + @Minio(accessKey = MINIO_ACCESS_KEY, secretKey = MINIO_SECRET_KEY) MinioAccess minioAccess) { + storageBase = minioAccess.s3BucketUri(BUCKET_URI_PREFIX); + endpoint = minioAccess.s3endpoint(); + } + + @Override + protected List allowedLocations() { + return List.of(storageBase.toString()); + } + + @Override + protected Optional endpoint() { + return Optional.of(endpoint); + } + + @Override + protected ImmutableMap.Builder clientFileIOProperties() { + return super.clientFileIOProperties() + // Grant direct access to the MinIO bucket; this FileIO instance does not + // use access delegation. + .put(StorageAccessProperty.AWS_KEY_ID.getPropertyName(), MINIO_ACCESS_KEY) + .put(StorageAccessProperty.AWS_SECRET_KEY.getPropertyName(), MINIO_SECRET_KEY); + } +} diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/common/CatalogHandler.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/common/CatalogHandler.java index 2d08ab2db0..7554a853b8 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/common/CatalogHandler.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/common/CatalogHandler.java @@ -239,6 +239,13 @@ protected void authorizeCreateTableLikeUnderNamespaceOperationOrThrow( initializeCatalog(); } + protected void authorizeRemoteSigningOperationOrThrow( + PolarisAuthorizableOperation op, TableIdentifier identifier) { + // Remote signing operations require the same access checks as creating a table-like + // entity under a namespace, so we can reuse the same authorization logic. + authorizeCreateTableLikeUnderNamespaceOperationOrThrow(op, identifier); + } + protected void authorizeBasicTableLikeOperationOrThrow( PolarisAuthorizableOperation op, PolarisEntitySubType subType, TableIdentifier identifier) { resolutionManifest = diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java index 7eec03595f..6fc14de4a4 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java @@ -28,8 +28,10 @@ import jakarta.annotation.Nonnull; import jakarta.annotation.Nullable; import jakarta.ws.rs.core.SecurityContext; +import jakarta.ws.rs.core.UriInfo; import java.io.Closeable; import java.io.IOException; +import java.net.URI; import java.net.URLEncoder; import java.nio.charset.Charset; import java.util.Arrays; @@ -118,15 +120,20 @@ import org.apache.polaris.core.persistence.resolver.ResolverFactory; import org.apache.polaris.core.persistence.resolver.ResolverPath; import org.apache.polaris.core.persistence.resolver.ResolverStatus; +import org.apache.polaris.core.rest.PolarisResourcePaths; import org.apache.polaris.core.storage.AccessConfig; import org.apache.polaris.core.storage.InMemoryStorageIntegration; import org.apache.polaris.core.storage.PolarisCredentialVendor; import org.apache.polaris.core.storage.PolarisStorageActions; import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo; import org.apache.polaris.core.storage.PolarisStorageIntegration; +import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider; import org.apache.polaris.core.storage.StorageLocation; import org.apache.polaris.core.storage.StorageUtil; +import org.apache.polaris.core.storage.aws.AwsCredentialsStorageIntegration; +import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo; import org.apache.polaris.core.storage.cache.StorageCredentialCache; +import org.apache.polaris.service.catalog.CatalogPrefixParser; import org.apache.polaris.service.catalog.SupportsNotifications; import org.apache.polaris.service.catalog.common.LocationUtils; import org.apache.polaris.service.catalog.io.FileIOFactory; @@ -149,7 +156,11 @@ /** Defines the relationship between PolarisEntities and Iceberg's business logic. */ public class IcebergCatalog extends BaseMetastoreViewCatalog - implements SupportsNamespaces, SupportsNotifications, Closeable, SupportsCredentialDelegation { + implements SupportsNamespaces, + SupportsNotifications, + Closeable, + SupportsCredentialDelegation, + SupportsRemoteSigning { private static final Logger LOGGER = LoggerFactory.getLogger(IcebergCatalog.class); private static final Joiner SLASH = Joiner.on("/"); @@ -177,6 +188,10 @@ public class IcebergCatalog extends BaseMetastoreViewCatalog private final TaskExecutor taskExecutor; private final SecurityContext securityContext; private final PolarisEventListener polarisEventListener; + private final PolarisStorageIntegrationProvider storageIntegrationProvider; + private final CatalogPrefixParser prefixParser; + private final UriInfo uriInfo; + private final AtomicBoolean loggedPrefixOverlapWarning = new AtomicBoolean(false); private String ioImplClassName; @@ -206,7 +221,10 @@ public IcebergCatalog( SecurityContext securityContext, TaskExecutor taskExecutor, FileIOFactory fileIOFactory, - PolarisEventListener polarisEventListener) { + PolarisEventListener polarisEventListener, + PolarisStorageIntegrationProvider storageIntegrationProvider, + CatalogPrefixParser prefixParser, + UriInfo uriInfo) { this.storageCredentialCache = storageCredentialCache; this.resolverFactory = resolverFactory; this.callContext = callContext; @@ -216,6 +234,9 @@ public IcebergCatalog( CatalogEntity.of(resolvedEntityView.getResolvedReferenceCatalogEntity().getRawLeafEntity()); this.securityContext = securityContext; this.taskExecutor = taskExecutor; + this.storageIntegrationProvider = storageIntegrationProvider; + this.prefixParser = prefixParser; + this.uriInfo = uriInfo; this.catalogId = catalogEntity.getId(); this.catalogName = catalogEntity.getName(); this.fileIOFactory = fileIOFactory; @@ -821,7 +842,7 @@ public boolean sendNotification( } @Override - public AccessConfig getAccessConfig( + public AccessConfig getAccessConfigForCredentialDelegation( TableIdentifier tableIdentifier, TableMetadata tableMetadata, Set storageActions) { @@ -834,13 +855,50 @@ public AccessConfig getAccessConfig( return AccessConfig.builder().build(); } return FileIOUtil.refreshAccessConfig( - callContext, - storageCredentialCache, - getCredentialVendor(), - tableIdentifier, - StorageUtil.getLocationsAllowedToBeAccessed(tableMetadata), - storageActions, - storageInfo.get()); + callContext, + storageCredentialCache, + getCredentialVendor(), + tableIdentifier, + StorageUtil.getLocationsAllowedToBeAccessed(tableMetadata), + storageActions, + storageInfo.get()) + .orElse(AccessConfig.EMPTY); + } + + @Override + public AccessConfig getAccessConfigForRemoteSigning(TableIdentifier tableIdentifier) { + + Optional configurationInfo = + findStorageInfo(tableIdentifier) + .map(PolarisEntity::getInternalPropertiesAsMap) + .map(info -> info.get(PolarisEntityConstants.getStorageConfigInfoPropertyName())) + .map(PolarisStorageConfigurationInfo::deserialize); + + if (configurationInfo.isEmpty()) { + LOGGER + .atWarn() + .addKeyValue("tableIdentifier", tableIdentifier) + .log("Remote signing: table entity has no storage configuration in its hierarchy"); + return AccessConfig.builder().build(); + } + + PolarisStorageIntegration storageIntegration = + storageIntegrationProvider.getStorageIntegrationForConfig(configurationInfo.get()); + + if (!(storageIntegration instanceof AwsCredentialsStorageIntegration)) { + LOGGER + .atWarn() + .addKeyValue("tableIdentifier", tableIdentifier) + .log("Table entity storage integration is not an AWS credentials storage integration"); + return AccessConfig.builder().build(); + } + + String prefix = prefixParser.catalogNameToPrefix(callContext.getRealmContext(), catalogName); + URI signerUri = uriInfo.getBaseUri().resolve("api/"); + String signerEndpoint = new PolarisResourcePaths(prefix).s3RemoteSigning(tableIdentifier); + + return ((AwsCredentialsStorageIntegration) storageIntegration) + .getRemoteSigningAccessConfig(signerUri, signerEndpoint); } private String buildPrefixedLocation(TableIdentifier tableIdentifier) { diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogAdapter.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogAdapter.java index 76401582a9..efdf368a3e 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogAdapter.java +++ b/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; @@ -36,6 +37,7 @@ import jakarta.ws.rs.core.SecurityContext; import java.util.EnumSet; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.function.Function; @@ -67,7 +69,7 @@ import org.apache.polaris.core.config.RealmConfig; import org.apache.polaris.core.context.CallContext; import org.apache.polaris.core.context.RealmContext; -import org.apache.polaris.core.entity.PolarisEntity; +import org.apache.polaris.core.entity.CatalogEntity; import org.apache.polaris.core.persistence.PolarisMetaStoreManager; import org.apache.polaris.core.persistence.ResolvedPolarisEntity; import org.apache.polaris.core.persistence.resolver.ResolutionManifestFactory; @@ -333,11 +335,13 @@ public Response updateProperties( catalog -> Response.ok(catalog.updateNamespaceProperties(ns, revisedRequest)).build()); } - private EnumSet parseAccessDelegationModes(String accessDelegationMode) { + private static Set parseAccessDelegationModes(String accessDelegationMode) { EnumSet 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; @@ -352,8 +356,7 @@ public Response createTable( RealmContext realmContext, SecurityContext securityContext) { validateIcebergProperties(realmConfig, createTableRequest.properties()); - EnumSet delegationModes = - parseAccessDelegationModes(accessDelegationMode); + Set delegationModes = parseAccessDelegationModes(accessDelegationMode); Namespace ns = decodeNamespace(namespace); return withCatalog( securityContext, @@ -364,7 +367,8 @@ public Response createTable( return Response.ok(catalog.createTableStaged(ns, createTableRequest)).build(); } else { return Response.ok( - catalog.createTableStagedWithWriteDelegation(ns, createTableRequest)) + catalog.createTableStagedWithWriteDelegation( + ns, createTableRequest, delegationModes)) .build(); } } else if (delegationModes.isEmpty()) { @@ -374,7 +378,8 @@ public Response createTable( .build(); } else { LoadTableResponse response = - catalog.createTableDirectWithWriteDelegation(ns, createTableRequest); + catalog.createTableDirectWithWriteDelegation( + ns, createTableRequest, delegationModes); return tryInsertETagHeader( Response.ok(response), response, namespace, createTableRequest.name()) .build(); @@ -407,8 +412,7 @@ public Response loadTable( String snapshots, RealmContext realmContext, SecurityContext securityContext) { - EnumSet delegationModes = - parseAccessDelegationModes(accessDelegationMode); + Set delegationModes = parseAccessDelegationModes(accessDelegationMode); Namespace ns = decodeNamespace(namespace); TableIdentifier tableIdentifier = TableIdentifier.of(ns, RESTUtil.decodeString(table)); @@ -432,7 +436,8 @@ public Response loadTable( } else { response = catalog - .loadTableWithAccessDelegationIfStale(tableIdentifier, ifNoneMatch, snapshots) + .loadTableWithAccessDelegationIfStale( + tableIdentifier, ifNoneMatch, delegationModes) .orElseThrow(() -> new WebApplicationException(Response.Status.NOT_MODIFIED)); } @@ -598,8 +603,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()) @@ -787,8 +791,9 @@ public Response getConfig( throw new NotFoundException("Unable to find warehouse %s", warehouse); } ResolvedPolarisEntity resolvedReferenceCatalog = resolver.getResolvedReferenceCatalog(); - Map properties = - PolarisEntity.of(resolvedReferenceCatalog.getEntity()).getPropertiesAsMap(); + CatalogEntity catalogEntity = + CatalogEntity.of(Objects.requireNonNull(resolvedReferenceCatalog).getEntity()); + Map properties = catalogEntity.getPropertiesAsMap(); String prefix = prefixParser.catalogNameToPrefix(realmContext, warehouse); return Response.ok( @@ -802,6 +807,9 @@ public Response getConfig( .addAll(COMMIT_ENDPOINT) .addAll(PolarisEndpoints.getSupportedGenericTableEndpoints(realmConfig)) .addAll(PolarisEndpoints.getSupportedPolicyEndpoints(realmConfig)) + .addAll( + PolarisEndpoints.getSupportedRemoteSigningEndpoints( + callContext.getRealmConfig(), catalogEntity)) .build()) .build()) .build(); diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java index b0cfc01b19..c55a219e2a 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java @@ -92,12 +92,14 @@ import org.apache.polaris.core.secrets.UserSecretsManager; import org.apache.polaris.core.storage.AccessConfig; import org.apache.polaris.core.storage.PolarisStorageActions; +import org.apache.polaris.service.catalog.AccessDelegationMode; import org.apache.polaris.service.catalog.SupportsNotifications; import org.apache.polaris.service.catalog.common.CatalogHandler; import org.apache.polaris.service.config.ReservedProperties; 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.s3.sign.S3RemoteSigningCatalogHandler; import org.apache.polaris.service.types.NotificationRequest; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -387,7 +389,7 @@ public LoadTableResponse createTableDirect(Namespace namespace, CreateTableReque * @return ETagged {@link LoadTableResponse} to uniquely identify the table metadata */ public LoadTableResponse createTableDirectWithWriteDelegation( - Namespace namespace, CreateTableRequest request) { + Namespace namespace, CreateTableRequest request, Set delegationModes) { PolarisAuthorizableOperation op = PolarisAuthorizableOperation.CREATE_TABLE_DIRECT_WITH_WRITE_DELEGATION; authorizeCreateTableLikeUnderNamespaceOperationOrThrow( @@ -426,7 +428,8 @@ public LoadTableResponse createTableDirectWithWriteDelegation( PolarisStorageActions.READ, PolarisStorageActions.WRITE, PolarisStorageActions.LIST), - SNAPSHOTS_ALL) + delegationModes, + catalog) .build(); } else if (table instanceof BaseMetadataTable) { // metadata tables are loaded on the client side, return NoSuchTableException for now @@ -494,7 +497,7 @@ public LoadTableResponse createTableStaged(Namespace namespace, CreateTableReque } public LoadTableResponse createTableStagedWithWriteDelegation( - Namespace namespace, CreateTableRequest request) { + Namespace namespace, CreateTableRequest request, Set delegationModes) { PolarisAuthorizableOperation op = PolarisAuthorizableOperation.CREATE_TABLE_STAGED_WITH_WRITE_DELEGATION; authorizeCreateTableLikeUnderNamespaceOperationOrThrow( @@ -508,7 +511,7 @@ public LoadTableResponse createTableStagedWithWriteDelegation( TableMetadata metadata = stageTableCreateHelper(namespace, request); return buildLoadTableResponseWithDelegationCredentials( - ident, metadata, Set.of(PolarisStorageActions.ALL), SNAPSHOTS_ALL) + ident, metadata, Set.of(PolarisStorageActions.ALL), delegationModes, catalog) .build(); } @@ -617,8 +620,8 @@ public Optional loadTableIfStale( } public LoadTableResponse loadTableWithAccessDelegation( - TableIdentifier tableIdentifier, String snapshots) { - return loadTableWithAccessDelegationIfStale(tableIdentifier, null, snapshots).get(); + TableIdentifier tableIdentifier, Set delegationModes) { + return loadTableWithAccessDelegationIfStale(tableIdentifier, null, delegationModes).get(); } /** @@ -627,12 +630,13 @@ public LoadTableResponse loadTableWithAccessDelegation( * * @param tableIdentifier The identifier of the table to load * @param ifNoneMatch set of entity-tags to check the metadata against for staleness - * @param snapshots * @return {@link Optional#empty()} if the ETag is current, an {@link Optional} containing the * load table response, otherwise */ public Optional loadTableWithAccessDelegationIfStale( - TableIdentifier tableIdentifier, IfNoneMatch ifNoneMatch, String snapshots) { + TableIdentifier tableIdentifier, + IfNoneMatch ifNoneMatch, + Set delegationModes) { // Here we have a single method that falls through multiple candidate // PolarisAuthorizableOperations because instead of identifying the desired operation up-front // and @@ -658,20 +662,18 @@ public Optional loadTableWithAccessDelegationIfStale( CatalogEntity catalogEntity = getResolvedCatalogEntity(); - LOGGER.info("Catalog type: {}", catalogEntity.getCatalogType()); - LOGGER.info( - "allow external catalog credential vending: {}", - realmConfig.getConfig( - FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING, catalogEntity)); - if (catalogEntity - .getCatalogType() - .equals(org.apache.polaris.core.admin.model.Catalog.TypeEnum.EXTERNAL) - && !realmConfig.getConfig( - FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING, catalogEntity)) { - throw new ForbiddenException( - "Access Delegation is not enabled for this catalog. Please consult applicable " - + "documentation for the catalog config property '%s' to enable this feature", - FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING.catalogConfig()); + if (delegationModes.contains(AccessDelegationMode.VENDED_CREDENTIALS)) { + boolean allowExternalCatalogCredentialVending = + realmConfig.getConfig( + FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING, catalogEntity); + LOGGER.info( + "allow external catalog credential vending: {}", allowExternalCatalogCredentialVending); + if (catalogEntity.isExternal() && !allowExternalCatalogCredentialVending) { + throw new ForbiddenException( + "Access Delegation is not enabled for this catalog. Please consult applicable " + + "documentation for the catalog config property '%s' to enable this feature", + FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING.catalogConfig()); + } } if (ifNoneMatch != null) { @@ -702,7 +704,7 @@ public Optional loadTableWithAccessDelegationIfStale( TableMetadata tableMetadata = baseTable.operations().current(); return Optional.of( buildLoadTableResponseWithDelegationCredentials( - tableIdentifier, tableMetadata, actionsRequested, snapshots) + tableIdentifier, tableMetadata, actionsRequested, delegationModes, catalogEntity) .build()); } else if (table instanceof BaseMetadataTable) { // metadata tables are loaded on the client side, return NoSuchTableException for now @@ -712,21 +714,48 @@ public Optional loadTableWithAccessDelegationIfStale( throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable"); } + private CatalogEntity getCatalogEntity() { + PolarisResolvedPathWrapper catalogPath = resolutionManifest.getResolvedReferenceCatalogEntity(); + callContext + .getPolarisCallContext() + .getDiagServices() + .checkNotNull(catalogPath, "No catalog available for loadTable request"); + CatalogEntity catalogEntity = CatalogEntity.of(catalogPath.getRawLeafEntity()); + LOGGER.info("Catalog type: {}", catalogEntity.getCatalogType()); + return catalogEntity; + } + private LoadTableResponse.Builder buildLoadTableResponseWithDelegationCredentials( TableIdentifier tableIdentifier, TableMetadata tableMetadata, Set actions, - String snapshots) { + Set delegationModes, + CatalogEntity catalogEntity) { LoadTableResponse.Builder responseBuilder = LoadTableResponse.builder().withTableMetadata(tableMetadata); - if (baseCatalog instanceof SupportsCredentialDelegation credentialDelegation) { + AccessConfig accessConfig = null; + if (baseCatalog instanceof SupportsCredentialDelegation credentialDelegation + && delegationModes.contains(AccessDelegationMode.VENDED_CREDENTIALS)) { LOGGER .atDebug() .addKeyValue("tableIdentifier", tableIdentifier) .addKeyValue("tableLocation", tableMetadata.location()) .log("Fetching client credentials for table"); - AccessConfig accessConfig = - credentialDelegation.getAccessConfig(tableIdentifier, tableMetadata, actions); + accessConfig = + credentialDelegation.getAccessConfigForCredentialDelegation( + tableIdentifier, tableMetadata, actions); + } else if (baseCatalog instanceof SupportsRemoteSigning remoteSigning + && delegationModes.contains(AccessDelegationMode.REMOTE_SIGNING)) { + S3RemoteSigningCatalogHandler.throwIfRemoteSigningNotEnabled( + callContext.getRealmConfig(), catalogEntity); + LOGGER + .atDebug() + .addKeyValue("tableIdentifier", tableIdentifier) + .addKeyValue("tableLocation", tableMetadata.location()) + .log("Enabling remote signing for table"); + accessConfig = remoteSigning.getAccessConfigForRemoteSigning(tableIdentifier); + } + if (accessConfig != null) { Map credentialConfig = accessConfig.credentials(); responseBuilder.addAllConfig(credentialConfig); responseBuilder.addAllConfig(accessConfig.extraProperties()); diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/SupportsCredentialDelegation.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/SupportsCredentialDelegation.java index 21ec380eb0..d792c96428 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/SupportsCredentialDelegation.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/SupportsCredentialDelegation.java @@ -32,7 +32,7 @@ * configuration. */ public interface SupportsCredentialDelegation { - AccessConfig getAccessConfig( + AccessConfig getAccessConfigForCredentialDelegation( TableIdentifier tableIdentifier, TableMetadata tableMetadata, Set storageActions); diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/SupportsRemoteSigning.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/SupportsRemoteSigning.java new file mode 100644 index 0000000000..41b268c072 --- /dev/null +++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/SupportsRemoteSigning.java @@ -0,0 +1,28 @@ +/* + * 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.catalog.iceberg; + +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.core.storage.AccessConfig; + +/** Adds support for remote signing of requests for Iceberg tables. */ +public interface SupportsRemoteSigning { + + AccessConfig getAccessConfigForRemoteSigning(TableIdentifier tableIdentifier); +} diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/DefaultFileIOFactory.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/DefaultFileIOFactory.java index d2c73e2684..75dba97fa4 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/DefaultFileIOFactory.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/DefaultFileIOFactory.java @@ -23,6 +23,7 @@ import jakarta.annotation.Nonnull; import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; +import java.time.Clock; import java.util.HashMap; import java.util.Map; import java.util.Optional; @@ -39,7 +40,9 @@ import org.apache.polaris.core.storage.AccessConfig; import org.apache.polaris.core.storage.PolarisCredentialVendor; import org.apache.polaris.core.storage.PolarisStorageActions; +import org.apache.polaris.core.storage.StorageAccessProperty; import org.apache.polaris.core.storage.cache.StorageCredentialCache; +import org.apache.polaris.service.storage.StorageConfiguration; /** * A default FileIO factory implementation for creating Iceberg {@link FileIO} instances with @@ -53,13 +56,19 @@ @Identifier("default") public class DefaultFileIOFactory implements FileIOFactory { + private final StorageConfiguration storageConfiguration; private final StorageCredentialCache storageCredentialCache; private final MetaStoreManagerFactory metaStoreManagerFactory; + private final Clock clock; @Inject public DefaultFileIOFactory( + StorageConfiguration storageConfiguration, StorageCredentialCache storageCredentialCache, - MetaStoreManagerFactory metaStoreManagerFactory) { + MetaStoreManagerFactory metaStoreManagerFactory, + Clock clock) { + this.storageConfiguration = storageConfiguration; + this.clock = clock; this.storageCredentialCache = storageCredentialCache; this.metaStoreManagerFactory = metaStoreManagerFactory; } @@ -82,7 +91,7 @@ public FileIO loadFileIO( Optional storageInfoEntity = FileIOUtil.findStorageInfoFromHierarchy(resolvedEntityPath); Optional accessConfig = - storageInfoEntity.map( + storageInfoEntity.flatMap( storageInfo -> FileIOUtil.refreshAccessConfig( callContext, @@ -101,6 +110,32 @@ public FileIO loadFileIO( properties.putAll(accessConfig.get().credentials()); properties.putAll(accessConfig.get().extraProperties()); properties.putAll(accessConfig.get().internalProperties()); + } else { + // If no subscoped creds were produced, use system-wide AWS or GCP credentials if available. + if (storageConfiguration.awsAccessKey().isPresent() + && storageConfiguration.awsSecretKey().isPresent()) { + // If no subscoped creds, use system-wide AWS credentials if available + properties.put( + StorageAccessProperty.AWS_KEY_ID.getPropertyName(), + storageConfiguration.awsAccessKey().get()); + properties.put( + StorageAccessProperty.AWS_SECRET_KEY.getPropertyName(), + storageConfiguration.awsSecretKey().get()); + } else if (storageConfiguration.gcpAccessToken().isPresent()) { + properties.put( + StorageAccessProperty.GCS_ACCESS_TOKEN.getPropertyName(), + storageConfiguration.gcpAccessToken().get()); + properties.put( + StorageAccessProperty.GCS_ACCESS_TOKEN_EXPIRES_AT.getPropertyName(), + String.valueOf( + clock + .instant() + .plus( + storageConfiguration + .gcpAccessTokenLifespan() + .orElse(StorageConfiguration.DEFAULT_TOKEN_LIFESPAN)) + .toEpochMilli())); + } } return loadFileIOInternal(ioImplClassName, properties); diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/FileIOUtil.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/FileIOUtil.java index c5ef12d784..27ce1bb94e 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/FileIOUtil.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/FileIOUtil.java @@ -63,7 +63,8 @@ public static Optional findStorageInfoFromHierarchy( } /** - * Refreshes or generates subscoped creds for accessing table storage based on the params. + * Refreshes or generates subscoped creds for accessing table storage based on the params. Returns + * empty if subscoped credentials are disabled. * *

Use cases: * @@ -73,8 +74,13 @@ public static Optional findStorageInfoFromHierarchy( *

  • In {@link DefaultFileIOFactory}, subscoped credentials are obtained to access the storage * and read/write metadata JSON files. * + * + *

    Note: when using S3 remote signing, this method is not called from {@link IcebergCatalog} + * since clients will use remote signing instead. However, it is still called from {@link + * DefaultFileIOFactory} to obtain subscoped credentials for the server itself, when it needs to + * access the storage to read/write metadata files. */ - public static AccessConfig refreshAccessConfig( + public static Optional refreshAccessConfig( CallContext callContext, StorageCredentialCache storageCredentialCache, PolarisCredentialVendor credentialVendor, @@ -92,7 +98,7 @@ public static AccessConfig refreshAccessConfig( .atDebug() .addKeyValue("tableIdentifier", tableIdentifier) .log("Skipping generation of subscoped creds for table"); - return AccessConfig.builder().build(); + return Optional.empty(); } boolean allowList = @@ -121,6 +127,6 @@ public static AccessConfig refreshAccessConfig( if (accessConfig.credentials().isEmpty()) { LOGGER.debug("No credentials found for table"); } - return accessConfig; + return Optional.of(accessConfig); } } diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/WasbTranslatingFileIOFactory.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/WasbTranslatingFileIOFactory.java index 048e19bb41..df142746cd 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/WasbTranslatingFileIOFactory.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/io/WasbTranslatingFileIOFactory.java @@ -22,6 +22,7 @@ import jakarta.annotation.Nonnull; import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; +import java.time.Clock; import java.util.Map; import java.util.Set; import org.apache.iceberg.catalog.TableIdentifier; @@ -31,6 +32,7 @@ import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper; import org.apache.polaris.core.storage.PolarisStorageActions; import org.apache.polaris.core.storage.cache.StorageCredentialCache; +import org.apache.polaris.service.storage.StorageConfiguration; /** A {@link FileIOFactory} that translates WASB paths to ABFS ones */ @ApplicationScoped @@ -41,10 +43,13 @@ public class WasbTranslatingFileIOFactory implements FileIOFactory { @Inject public WasbTranslatingFileIOFactory( + StorageConfiguration storageConfiguration, StorageCredentialCache storageCredentialCache, - MetaStoreManagerFactory metaStoreManagerFactory) { + MetaStoreManagerFactory metaStoreManagerFactory, + Clock clock) { defaultFileIOFactory = - new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory); + new DefaultFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock); } @Override diff --git a/runtime/service/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java b/runtime/service/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java index 31360ae31d..8a832cf1a0 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java @@ -21,6 +21,7 @@ import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; import jakarta.ws.rs.core.SecurityContext; +import jakarta.ws.rs.core.UriInfo; import java.util.HashMap; import java.util.Map; import org.apache.iceberg.CatalogProperties; @@ -32,7 +33,9 @@ import org.apache.polaris.core.persistence.MetaStoreManagerFactory; import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest; import org.apache.polaris.core.persistence.resolver.ResolverFactory; +import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider; import org.apache.polaris.core.storage.cache.StorageCredentialCache; +import org.apache.polaris.service.catalog.CatalogPrefixParser; import org.apache.polaris.service.catalog.iceberg.IcebergCatalog; import org.apache.polaris.service.catalog.io.FileIOFactory; import org.apache.polaris.service.events.PolarisEventListener; @@ -51,6 +54,9 @@ public class PolarisCallContextCatalogFactory implements CallContextCatalogFacto private final ResolverFactory resolverFactory; private final MetaStoreManagerFactory metaStoreManagerFactory; private final PolarisEventListener polarisEventListener; + private final PolarisStorageIntegrationProvider storageIntegrationProvider; + private final CatalogPrefixParser prefixParser; + private final UriInfo uriInfo; @Inject public PolarisCallContextCatalogFactory( @@ -59,13 +65,19 @@ public PolarisCallContextCatalogFactory( MetaStoreManagerFactory metaStoreManagerFactory, TaskExecutor taskExecutor, FileIOFactory fileIOFactory, - PolarisEventListener polarisEventListener) { + PolarisEventListener polarisEventListener, + PolarisStorageIntegrationProvider storageIntegrationProvider, + CatalogPrefixParser prefixParser, + UriInfo uriInfo) { this.storageCredentialCache = storageCredentialCache; this.resolverFactory = resolverFactory; this.metaStoreManagerFactory = metaStoreManagerFactory; this.taskExecutor = taskExecutor; this.fileIOFactory = fileIOFactory; this.polarisEventListener = polarisEventListener; + this.storageIntegrationProvider = storageIntegrationProvider; + this.prefixParser = prefixParser; + this.uriInfo = uriInfo; } @Override @@ -92,7 +104,10 @@ public Catalog createCallContextCatalog( securityContext, taskExecutor, fileIOFactory, - polarisEventListener); + polarisEventListener, + storageIntegrationProvider, + prefixParser, + uriInfo); CatalogEntity catalog = CatalogEntity.of(baseCatalogEntity); Map catalogProperties = new HashMap<>(catalog.getPropertiesAsMap()); diff --git a/runtime/service/src/main/java/org/apache/polaris/service/storage/PolarisStorageIntegrationProviderImpl.java b/runtime/service/src/main/java/org/apache/polaris/service/storage/PolarisStorageIntegrationProviderImpl.java index e07bdd0823..5ee6d64e19 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/storage/PolarisStorageIntegrationProviderImpl.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/storage/PolarisStorageIntegrationProviderImpl.java @@ -59,7 +59,7 @@ public PolarisStorageIntegrationProviderImpl( StorageConfiguration storageConfiguration, StsClientProvider stsClientProvider, Clock clock) { this( stsClientProvider, - Optional.ofNullable(storageConfiguration.stsCredentials()), + Optional.ofNullable(storageConfiguration.awsSystemCredentials()), storageConfiguration.gcpCredentialsSupplier(clock)); } diff --git a/runtime/service/src/main/java/org/apache/polaris/service/storage/StorageConfiguration.java b/runtime/service/src/main/java/org/apache/polaris/service/storage/StorageConfiguration.java index f92c45416a..aba36dc79f 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/storage/StorageConfiguration.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/storage/StorageConfiguration.java @@ -36,7 +36,6 @@ import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider; import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; import software.amazon.awssdk.services.sts.StsClient; -import software.amazon.awssdk.services.sts.StsClientBuilder; @ConfigMapping(prefix = "polaris.storage") public interface StorageConfiguration extends S3AccessConfig { @@ -72,21 +71,19 @@ public interface StorageConfiguration extends S3AccessConfig { Optional gcpAccessTokenLifespan(); default Supplier stsClientSupplier() { - return stsClientSupplier(true); - } - - default Supplier stsClientSupplier(boolean withCredentials) { return Suppliers.memoize( - () -> { - StsClientBuilder stsClientBuilder = StsClient.builder(); - if (withCredentials) { - stsClientBuilder.credentialsProvider(stsCredentials()); - } - return stsClientBuilder.build(); - }); + () -> StsClient.builder().credentialsProvider(awsSystemCredentials()).build()); } - default AwsCredentialsProvider stsCredentials() { + /** + * Returns an {@link AwsCredentialsProvider} that provides system-wide AWS credentials. If both + * access key and secret key are present, it uses them directly; otherwise, it uses the default + * credentials provider chain. + * + *

    The returned provider is not meant to be vended directly to clients, but rather used with + * STS, unless credential subscoping is disabled. + */ + default AwsCredentialsProvider awsSystemCredentials() { if (awsAccessKey().isPresent() && awsSecretKey().isPresent()) { LoggerFactory.getLogger(StorageConfiguration.class) .warn("Using hard-coded AWS credentials - this is not recommended for production"); diff --git a/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogAdapter.java b/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogAdapter.java new file mode 100644 index 0000000000..b17db90898 --- /dev/null +++ b/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogAdapter.java @@ -0,0 +1,100 @@ +/* + * 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.storage.s3.sign; + +import jakarta.enterprise.context.RequestScoped; +import jakarta.inject.Inject; +import jakarta.ws.rs.core.Response; +import jakarta.ws.rs.core.SecurityContext; +import java.util.function.Function; +import org.apache.iceberg.aws.s3.signer.S3SignResponse; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.rest.RESTUtil; +import org.apache.polaris.core.auth.PolarisAuthorizer; +import org.apache.polaris.core.context.CallContext; +import org.apache.polaris.core.context.RealmContext; +import org.apache.polaris.core.persistence.resolver.ResolutionManifestFactory; +import org.apache.polaris.service.catalog.CatalogPrefixParser; +import org.apache.polaris.service.catalog.common.CatalogAdapter; +import org.apache.polaris.service.s3.sign.api.IcebergRestS3SignerApiService; +import org.apache.polaris.service.s3.sign.model.PolarisS3SignRequest; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** A bridge between {@link IcebergRestS3SignerApiService} and {@link CatalogAdapter}. */ +@RequestScoped +public class S3RemoteSigningCatalogAdapter + implements IcebergRestS3SignerApiService, CatalogAdapter { + + private static final Logger LOGGER = LoggerFactory.getLogger(S3RemoteSigningCatalogAdapter.class); + + @Inject CallContext callContext; + @Inject ResolutionManifestFactory resolutionManifestFactory; + @Inject PolarisAuthorizer polarisAuthorizer; + @Inject CatalogPrefixParser prefixParser; + @Inject S3RequestSigner s3RequestSigner; + + /** + * Execute operations on a catalog wrapper and ensure we close the BaseCatalog afterward. This + * will typically ensure the underlying FileIO is closed. + */ + private Response withCatalog( + SecurityContext securityContext, + String prefix, + Function action) { + validatePrincipal(securityContext); + String catalogName = prefixParser.prefixToCatalogName(callContext.getRealmContext(), prefix); + try (S3RemoteSigningCatalogHandler handler = + new S3RemoteSigningCatalogHandler( + callContext, + resolutionManifestFactory, + securityContext, + catalogName, + polarisAuthorizer, + s3RequestSigner)) { + return action.apply(handler); + } catch (RuntimeException e) { + LOGGER.debug("RuntimeException while operating on catalog. Propagating to caller.", e); + throw e; + } catch (Exception e) { + LOGGER.error("Error while operating on catalog", e); + throw new RuntimeException(e); + } + } + + @Override + public Response signS3Request( + String prefix, + String namespace, + String table, + PolarisS3SignRequest s3SignRequest, + RealmContext realmContext, + SecurityContext securityContext) { + Namespace ns = decodeNamespace(namespace); + TableIdentifier tableIdentifier = TableIdentifier.of(ns, RESTUtil.decodeString(table)); + return withCatalog( + securityContext, + prefix, + catalog -> { + S3SignResponse response = catalog.signS3Request(s3SignRequest, tableIdentifier); + return Response.status(Response.Status.OK).entity(response).build(); + }); + } +} diff --git a/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogHandler.java b/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogHandler.java new file mode 100644 index 0000000000..e80aed1d6e --- /dev/null +++ b/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogHandler.java @@ -0,0 +1,119 @@ +/* + * 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.storage.s3.sign; + +import jakarta.ws.rs.core.SecurityContext; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.ForbiddenException; +import org.apache.polaris.core.auth.PolarisAuthorizableOperation; +import org.apache.polaris.core.auth.PolarisAuthorizer; +import org.apache.polaris.core.config.FeatureConfiguration; +import org.apache.polaris.core.config.RealmConfig; +import org.apache.polaris.core.context.CallContext; +import org.apache.polaris.core.entity.CatalogEntity; +import org.apache.polaris.core.persistence.resolver.ResolutionManifestFactory; +import org.apache.polaris.service.catalog.common.CatalogHandler; +import org.apache.polaris.service.s3.sign.model.PolarisS3SignRequest; +import org.apache.polaris.service.s3.sign.model.PolarisS3SignResponse; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class S3RemoteSigningCatalogHandler extends CatalogHandler implements AutoCloseable { + + private static final Logger LOGGER = LoggerFactory.getLogger(S3RemoteSigningCatalogHandler.class); + + private final S3RequestSigner s3RequestSigner; + + private CatalogEntity catalogEntity; + + public S3RemoteSigningCatalogHandler( + CallContext callContext, + ResolutionManifestFactory resolutionManifestFactory, + SecurityContext securityContext, + String catalogName, + PolarisAuthorizer authorizer, + S3RequestSigner s3RequestSigner) { + super( + callContext, + resolutionManifestFactory, + securityContext, + catalogName, + authorizer, + // external catalogs are not supported for S3 remote signing + null, + null); + this.s3RequestSigner = s3RequestSigner; + } + + @Override + protected void initializeCatalog() { + catalogEntity = + CatalogEntity.of(resolutionManifest.getResolvedReferenceCatalogEntity().getRawLeafEntity()); + if (catalogEntity.isExternal()) { + throw new ForbiddenException("Cannot use S3 remote signing with federated catalogs."); + } + // no need to materialize the catalog here, as we only need the catalog entity + } + + public PolarisS3SignResponse signS3Request( + PolarisS3SignRequest s3SignRequest, TableIdentifier tableIdentifier) { + + LOGGER.debug("Requesting s3 signing for {}: {}", tableIdentifier, s3SignRequest); + + PolarisAuthorizableOperation authzOp = + s3SignRequest.write() + ? PolarisAuthorizableOperation.SIGN_S3_WRITE_REQUEST + : PolarisAuthorizableOperation.SIGN_S3_READ_REQUEST; + + authorizeRemoteSigningOperationOrThrow(authzOp, tableIdentifier); + + // Must be done after the authorization check, as the auth check creates the catalog entity + throwIfRemoteSigningNotEnabled(callContext.getRealmConfig(), catalogEntity); + + // TODO S3 location access checks + // - The requested S3 location should be within the catalog's allowed read or write locations. + // - The requested S3 location is not allowed to be: + // - When writing: the table's metadata location + // - When reading: none. + + PolarisS3SignResponse s3SignResponse = s3RequestSigner.signRequest(s3SignRequest); + LOGGER.debug("S3 signing response: {}", s3SignResponse); + + return s3SignResponse; + } + + public static void throwIfRemoteSigningNotEnabled( + RealmConfig realmConfig, CatalogEntity catalogEntity) { + if (catalogEntity.isExternal()) { + throw new ForbiddenException("Remote signing is not enabled for external catalogs."); + } + boolean remoteSigningEnabled = + realmConfig.getConfig(FeatureConfiguration.REMOTE_SIGNING_ENABLED, catalogEntity); + if (!remoteSigningEnabled) { + throw new ForbiddenException( + "Remote signing is not enabled for this catalog. To enable this feature, set the Polaris configuration %s " + + "or the catalog configuration %s.", + FeatureConfiguration.REMOTE_SIGNING_ENABLED.key(), + FeatureConfiguration.REMOTE_SIGNING_ENABLED.catalogConfig()); + } + } + + @Override + public void close() {} +} diff --git a/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RequestSigner.java b/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RequestSigner.java new file mode 100644 index 0000000000..63175338be --- /dev/null +++ b/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RequestSigner.java @@ -0,0 +1,30 @@ +/* + * 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.storage.s3.sign; + +import org.apache.iceberg.aws.s3.signer.S3SignRequest; +import org.apache.polaris.service.s3.sign.model.PolarisS3SignResponse; + +/** Interface for signing S3 requests. */ +public interface S3RequestSigner { + + /** Signs an S3 request. */ + PolarisS3SignResponse signRequest(S3SignRequest signingRequest); +} diff --git a/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RequestSignerImpl.java b/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RequestSignerImpl.java new file mode 100644 index 0000000000..3fdd0800f0 --- /dev/null +++ b/runtime/service/src/main/java/org/apache/polaris/service/storage/s3/sign/S3RequestSignerImpl.java @@ -0,0 +1,84 @@ +/* + * 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.storage.s3.sign; + +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.net.URI; +import org.apache.iceberg.aws.s3.signer.S3SignRequest; +import org.apache.polaris.service.s3.sign.model.ImmutablePolarisS3SignResponse; +import org.apache.polaris.service.s3.sign.model.PolarisS3SignResponse; +import org.apache.polaris.service.storage.StorageConfiguration; +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.http.ContentStreamProvider; +import software.amazon.awssdk.http.SdkHttpFullRequest; +import software.amazon.awssdk.http.SdkHttpMethod; +import software.amazon.awssdk.http.SdkHttpRequest; +import software.amazon.awssdk.http.auth.aws.signer.AwsV4HttpSigner; +import software.amazon.awssdk.http.auth.spi.signer.SignRequest; +import software.amazon.awssdk.http.auth.spi.signer.SignedRequest; +import software.amazon.awssdk.services.s3.S3Client; + +@ApplicationScoped +class S3RequestSignerImpl implements S3RequestSigner { + + private final AwsV4HttpSigner signer = AwsV4HttpSigner.create(); + + @Inject StorageConfiguration storageConfiguration; + + @Override + public PolarisS3SignResponse signRequest(S3SignRequest signingRequest) { + + URI uri = signingRequest.uri(); + SdkHttpMethod method = SdkHttpMethod.valueOf(signingRequest.method()); + + SdkHttpFullRequest.Builder requestToSign = + SdkHttpFullRequest.builder() + .uri(uri) + .protocol(uri.getScheme()) + .method(method) + .headers(signingRequest.headers()); + + AwsCredentials credentials = storageConfiguration.awsSystemCredentials().resolveCredentials(); + + SignRequest.Builder signRequest = + SignRequest.builder(credentials) + .request(requestToSign.build()) + .putProperty(AwsV4HttpSigner.REGION_NAME, signingRequest.region()) + .putProperty(AwsV4HttpSigner.SERVICE_SIGNING_NAME, S3Client.SERVICE_NAME) + .putProperty(AwsV4HttpSigner.DOUBLE_URL_ENCODE, false) + .putProperty(AwsV4HttpSigner.NORMALIZE_PATH, false) + .putProperty(AwsV4HttpSigner.CHUNK_ENCODING_ENABLED, false) + .putProperty(AwsV4HttpSigner.PAYLOAD_SIGNING_ENABLED, false); + + String body = signingRequest.body(); + if (body != null) { + signRequest.payload(ContentStreamProvider.fromUtf8String(body)); + } + + SignedRequest signed = signer.sign(signRequest.build()); + SdkHttpRequest signedRequest = signed.request(); + + return ImmutablePolarisS3SignResponse.builder() + .uri(signedRequest.getUri()) + .headers(signedRequest.headers()) + .build(); + } +} diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java index 2717ee71c9..3f77106804 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java @@ -29,7 +29,9 @@ import jakarta.enterprise.inject.Alternative; import jakarta.inject.Inject; import jakarta.ws.rs.core.SecurityContext; +import jakarta.ws.rs.core.UriInfo; import java.io.IOException; +import java.time.Clock; import java.util.Date; import java.util.List; import java.util.Map; @@ -76,7 +78,9 @@ import org.apache.polaris.core.policy.PredefinedPolicyTypes; import org.apache.polaris.core.secrets.UserSecretsManager; import org.apache.polaris.core.secrets.UserSecretsManagerFactory; +import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider; import org.apache.polaris.core.storage.cache.StorageCredentialCache; +import org.apache.polaris.service.catalog.CatalogPrefixParser; import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView; import org.apache.polaris.service.catalog.Profiles; import org.apache.polaris.service.catalog.generic.PolarisGenericTableCatalog; @@ -89,6 +93,7 @@ import org.apache.polaris.service.context.catalog.PolarisCallContextCatalogFactory; import org.apache.polaris.service.events.PolarisEventListener; import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl; +import org.apache.polaris.service.storage.StorageConfiguration; import org.apache.polaris.service.task.TaskExecutor; import org.apache.polaris.service.types.PolicyIdentifier; import org.assertj.core.api.Assertions; @@ -119,6 +124,7 @@ public Map getConfigOverrides() { "polaris.features.\"ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING\"", "true") .put("polaris.features.\"DROP_WITH_PURGE_ENABLED\"", "true") + .put("polaris.features.\"REMOTE_SIGNING_ENABLED\"", "true") .build(); } } @@ -191,7 +197,11 @@ public Map getConfigOverrides() { @Inject protected CatalogHandlerUtils catalogHandlerUtils; @Inject protected PolarisConfigurationStore configurationStore; @Inject protected StorageCredentialCache storageCredentialCache; + @Inject protected StorageConfiguration storageConfiguration; @Inject protected ResolverFactory resolverFactory; + @Inject protected PolarisStorageIntegrationProvider storageIntegrationProvider; + @Inject protected CatalogPrefixParser prefixParser; + @Inject protected Clock clock; protected IcebergCatalog baseCatalog; protected PolarisGenericTableCatalog genericTableCatalog; @@ -461,7 +471,10 @@ private void initBaseCatalog() { securityContext, Mockito.mock(), fileIOFactory, - polarisEventListener); + polarisEventListener, + storageIntegrationProvider, + prefixParser, + Mockito.mock()); this.baseCatalog.initialize( CATALOG_NAME, ImmutableMap.of( @@ -479,7 +492,7 @@ public static class TestPolarisCallContextCatalogFactory @SuppressWarnings("unused") // Required by CDI protected TestPolarisCallContextCatalogFactory() { - this(null, null, null, null, null, null); + this(null, null, null, null, null, null, null, null, null); } @Inject @@ -489,14 +502,20 @@ public TestPolarisCallContextCatalogFactory( MetaStoreManagerFactory metaStoreManagerFactory, TaskExecutor taskExecutor, FileIOFactory fileIOFactory, - PolarisEventListener polarisEventListener) { + PolarisEventListener polarisEventListener, + PolarisStorageIntegrationProvider storageIntegrationProvider, + CatalogPrefixParser prefixParser, + UriInfo uriInfo) { super( storageCredentialCache, resolverFactory, metaStoreManagerFactory, taskExecutor, fileIOFactory, - polarisEventListener); + polarisEventListener, + storageIntegrationProvider, + prefixParser, + uriInfo); } @Override diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisS3InteroperabilityTest.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisS3InteroperabilityTest.java index d7b4abc8c9..63ef0a870f 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisS3InteroperabilityTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisS3InteroperabilityTest.java @@ -73,7 +73,7 @@ private static String makeTableLocation( public PolarisS3InteroperabilityTest() { TestServices.FileIOFactorySupplier fileIOFactorySupplier = - (storageCredentialCache, metaStoreManagerFactory) -> + (storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock) -> (FileIOFactory) (callContext, ioImplClassName, diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractIcebergCatalogTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractIcebergCatalogTest.java index 38c4db7e5f..280e57005d 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractIcebergCatalogTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractIcebergCatalogTest.java @@ -145,6 +145,7 @@ import org.apache.polaris.service.exception.FakeAzureHttpResponse; import org.apache.polaris.service.exception.IcebergExceptionMapper; import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl; +import org.apache.polaris.service.storage.StorageConfiguration; import org.apache.polaris.service.task.TableCleanupTaskHandler; import org.apache.polaris.service.task.TaskExecutor; import org.apache.polaris.service.task.TaskFileIOSupplier; @@ -225,11 +226,13 @@ public Map getConfigOverrides() { @Inject Clock clock; @Inject MetaStoreManagerFactory metaStoreManagerFactory; @Inject PolarisConfigurationStore configurationStore; + @Inject StorageConfiguration storageConfiguration; @Inject StorageCredentialCache storageCredentialCache; @Inject PolarisStorageIntegrationProvider storageIntegrationProvider; @Inject UserSecretsManagerFactory userSecretsManagerFactory; @Inject PolarisDiagnostics diagServices; @Inject PolarisEventListener polarisEventListener; + @Inject CatalogPrefixParser prefixParser; private IcebergCatalog catalog; private String realmName; @@ -344,7 +347,9 @@ public void before(TestInfo testInfo) { .build() .asCatalog())); - this.fileIOFactory = new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory); + this.fileIOFactory = + new DefaultFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock); StsClient stsClient = Mockito.mock(StsClient.class); when(stsClient.assumeRole(isA(AssumeRoleRequest.class))) @@ -446,7 +451,10 @@ protected IcebergCatalog newIcebergCatalog( securityContext, taskExecutor, fileIOFactory, - polarisEventListener); + polarisEventListener, + storageIntegrationProvider, + prefixParser, + Mockito.mock()); } @Test @@ -988,7 +996,9 @@ public void testValidateNotificationFailToCreateFileIO() { final String tableLocation = "s3://externally-owned-bucket/validate_table/"; final String tableMetadataLocation = tableLocation + "metadata/"; FileIOFactory fileIOFactory = - spy(new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory)); + spy( + new DefaultFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock)); IcebergCatalog catalog = newIcebergCatalog(catalog().name(), metaStoreManager, fileIOFactory); catalog.initialize( CATALOG_NAME, @@ -1839,7 +1849,8 @@ public void testDropTableWithPurge() { .containsEntry(StorageAccessProperty.AWS_TOKEN.getPropertyName(), SESSION_TOKEN); FileIO fileIO = new TaskFileIOSupplier( - new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory)) + new DefaultFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock)) .apply(taskEntity, polarisContext); Assertions.assertThat(fileIO).isNotNull().isInstanceOf(ExceptionMappingFileIO.class); Assertions.assertThat(((ExceptionMappingFileIO) fileIO).getInnerIo()) @@ -1966,7 +1977,8 @@ static Stream testRetriableException() { @Test public void testFileIOWrapper() { MeasuredFileIOFactory measured = - new MeasuredFileIOFactory(storageCredentialCache, metaStoreManagerFactory); + new MeasuredFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock); IcebergCatalog catalog = newIcebergCatalog(CATALOG_NAME, metaStoreManager, measured); catalog.initialize( CATALOG_NAME, diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractIcebergCatalogViewTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractIcebergCatalogViewTest.java index 92fff50cf4..7e9b43e9cf 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractIcebergCatalogViewTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractIcebergCatalogViewTest.java @@ -28,6 +28,7 @@ import java.lang.reflect.Field; import java.lang.reflect.Method; import java.nio.file.Path; +import java.time.Clock; import java.util.List; import java.util.Map; import java.util.Set; @@ -55,6 +56,7 @@ import org.apache.polaris.core.persistence.resolver.ResolverFactory; import org.apache.polaris.core.secrets.UserSecretsManager; import org.apache.polaris.core.secrets.UserSecretsManagerFactory; +import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider; import org.apache.polaris.core.storage.cache.StorageCredentialCache; import org.apache.polaris.service.admin.PolarisAdminService; import org.apache.polaris.service.catalog.iceberg.IcebergCatalog; @@ -68,6 +70,7 @@ import org.apache.polaris.service.events.PolarisEventListener; import org.apache.polaris.service.events.TestPolarisEventListener; import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl; +import org.apache.polaris.service.storage.StorageConfiguration; import org.apache.polaris.service.test.TestData; import org.assertj.core.api.Assertions; import org.assertj.core.api.Assumptions; @@ -110,11 +113,15 @@ public Map getConfigOverrides() { @Inject MetaStoreManagerFactory metaStoreManagerFactory; @Inject UserSecretsManagerFactory userSecretsManagerFactory; @Inject PolarisConfigurationStore configurationStore; + @Inject StorageConfiguration storageConfiguration; @Inject StorageCredentialCache storageCredentialCache; @Inject PolarisDiagnostics diagServices; @Inject PolarisEventListener polarisEventListener; @Inject ResolverFactory resolverFactory; @Inject ResolutionManifestFactory resolutionManifestFactory; + @Inject PolarisStorageIntegrationProvider storageIntegrationProvider; + @Inject CatalogPrefixParser prefixParser; + @Inject Clock clock; private IcebergCatalog catalog; @@ -207,7 +214,8 @@ public void before(TestInfo testInfo) { new PolarisPassthroughResolutionView( polarisContext, resolutionManifestFactory, securityContext, CATALOG_NAME); FileIOFactory fileIOFactory = - new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory); + new DefaultFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock); testPolarisEventListener = (TestPolarisEventListener) polarisEventListener; this.catalog = @@ -220,7 +228,10 @@ public void before(TestInfo testInfo) { securityContext, Mockito.mock(), fileIOFactory, - polarisEventListener); + polarisEventListener, + storageIntegrationProvider, + prefixParser, + Mockito.mock()); Map properties = ImmutableMap.builder() .put(CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO") diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractPolarisGenericTableCatalogTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractPolarisGenericTableCatalogTest.java index af7556618d..c2a2a70440 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractPolarisGenericTableCatalogTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractPolarisGenericTableCatalogTest.java @@ -28,6 +28,7 @@ import jakarta.ws.rs.core.SecurityContext; import java.io.IOException; import java.lang.reflect.Method; +import java.time.Clock; import java.util.List; import java.util.Map; import java.util.Set; @@ -71,6 +72,7 @@ import org.apache.polaris.service.config.ReservedProperties; import org.apache.polaris.service.events.NoOpPolarisEventListener; import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl; +import org.apache.polaris.service.storage.StorageConfiguration; import org.apache.polaris.service.task.TaskExecutor; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.AfterEach; @@ -99,11 +101,14 @@ public abstract class AbstractPolarisGenericTableCatalogTest { @Inject MetaStoreManagerFactory metaStoreManagerFactory; @Inject UserSecretsManagerFactory userSecretsManagerFactory; @Inject PolarisConfigurationStore configurationStore; + @Inject StorageConfiguration storageConfiguration; @Inject StorageCredentialCache storageCredentialCache; @Inject PolarisStorageIntegrationProvider storageIntegrationProvider; @Inject PolarisDiagnostics diagServices; @Inject ResolverFactory resolverFactory; @Inject ResolutionManifestFactory resolutionManifestFactory; + @Inject CatalogPrefixParser prefixParser; + @Inject Clock clock; private PolarisGenericTableCatalog genericTableCatalog; private IcebergCatalog icebergCatalog; @@ -208,7 +213,9 @@ public void before(TestInfo testInfo) { new PolarisPassthroughResolutionView( polarisContext, resolutionManifestFactory, securityContext, CATALOG_NAME); TaskExecutor taskExecutor = Mockito.mock(); - this.fileIOFactory = new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory); + this.fileIOFactory = + new DefaultFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock); StsClient stsClient = Mockito.mock(StsClient.class); when(stsClient.assumeRole(isA(AssumeRoleRequest.class))) @@ -243,7 +250,10 @@ public void before(TestInfo testInfo) { securityContext, taskExecutor, fileIOFactory, - new NoOpPolarisEventListener()); + new NoOpPolarisEventListener(), + storageIntegrationProvider, + prefixParser, + Mockito.mock()); this.icebergCatalog.initialize( CATALOG_NAME, ImmutableMap.of( diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractPolicyCatalogTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractPolicyCatalogTest.java index 2e34acac93..0e6c386b30 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractPolicyCatalogTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractPolicyCatalogTest.java @@ -34,6 +34,7 @@ import jakarta.ws.rs.core.SecurityContext; import java.io.IOException; import java.lang.reflect.Method; +import java.time.Clock; import java.util.Arrays; import java.util.List; import java.util.Set; @@ -83,6 +84,7 @@ import org.apache.polaris.service.config.ReservedProperties; import org.apache.polaris.service.events.NoOpPolarisEventListener; import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl; +import org.apache.polaris.service.storage.StorageConfiguration; import org.apache.polaris.service.task.TaskExecutor; import org.apache.polaris.service.types.ApplicablePolicy; import org.apache.polaris.service.types.Policy; @@ -125,11 +127,14 @@ public abstract class AbstractPolicyCatalogTest { @Inject MetaStoreManagerFactory metaStoreManagerFactory; @Inject UserSecretsManagerFactory userSecretsManagerFactory; @Inject PolarisConfigurationStore configurationStore; + @Inject StorageConfiguration storageConfiguration; @Inject StorageCredentialCache storageCredentialCache; @Inject PolarisStorageIntegrationProvider storageIntegrationProvider; @Inject PolarisDiagnostics diagServices; @Inject ResolverFactory resolverFactory; @Inject ResolutionManifestFactory resolutionManifestFactory; + @Inject CatalogPrefixParser prefixParser; + @Inject Clock clock; private PolicyCatalog policyCatalog; private IcebergCatalog icebergCatalog; @@ -227,7 +232,9 @@ public void before(TestInfo testInfo) { new PolarisPassthroughResolutionView( polarisContext, resolutionManifestFactory, securityContext, CATALOG_NAME); TaskExecutor taskExecutor = Mockito.mock(); - this.fileIOFactory = new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory); + this.fileIOFactory = + new DefaultFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock); StsClient stsClient = Mockito.mock(StsClient.class); when(stsClient.assumeRole(isA(AssumeRoleRequest.class))) @@ -260,7 +267,10 @@ public void before(TestInfo testInfo) { securityContext, taskExecutor, fileIOFactory, - new NoOpPolarisEventListener()); + new NoOpPolarisEventListener(), + storageIntegrationProvider, + prefixParser, + Mockito.mock()); this.icebergCatalog.initialize( CATALOG_NAME, ImmutableMap.of( diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/IcebergCatalogHandlerAuthzTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/IcebergCatalogHandlerAuthzTest.java index a3ab18e3f4..4634699363 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/IcebergCatalogHandlerAuthzTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/IcebergCatalogHandlerAuthzTest.java @@ -616,7 +616,10 @@ public void testCreateTableDirectWithWriteDelegationAllSufficientPrivileges() { Set.of(PolarisPrivilege.CATALOG_MANAGE_CONTENT)), () -> { newWrapper(Set.of(PRINCIPAL_ROLE1)) - .createTableDirectWithWriteDelegation(NS2, createDirectWithWriteDelegationRequest); + .createTableDirectWithWriteDelegation( + NS2, + createDirectWithWriteDelegationRequest, + Set.of(AccessDelegationMode.VENDED_CREDENTIALS)); }, () -> { newWrapper(Set.of(PRINCIPAL_ROLE2)).dropTableWithPurge(newtable); @@ -646,7 +649,10 @@ public void testCreateTableDirectWithWriteDelegationInsufficientPermissions() { PolarisPrivilege.TABLE_LIST), () -> { newWrapper(Set.of(PRINCIPAL_ROLE1)) - .createTableDirectWithWriteDelegation(NS2, createDirectWithWriteDelegationRequest); + .createTableDirectWithWriteDelegation( + NS2, + createDirectWithWriteDelegationRequest, + Set.of(AccessDelegationMode.VENDED_CREDENTIALS)); }); } @@ -719,7 +725,10 @@ public void testCreateTableStagedWithWriteDelegationAllSufficientPrivileges() { Set.of(PolarisPrivilege.CATALOG_MANAGE_CONTENT)), () -> { newWrapper(Set.of(PRINCIPAL_ROLE1)) - .createTableStagedWithWriteDelegation(NS2, createStagedWithWriteDelegationRequest); + .createTableStagedWithWriteDelegation( + NS2, + createStagedWithWriteDelegationRequest, + Set.of(AccessDelegationMode.VENDED_CREDENTIALS)); }, // createTableStagedWithWriteDelegation doesn't actually commit any metadata null, @@ -748,7 +757,10 @@ public void testCreateTableStagedWithWriteDelegationInsufficientPermissions() { PolarisPrivilege.TABLE_LIST), () -> { newWrapper(Set.of(PRINCIPAL_ROLE1)) - .createTableStagedWithWriteDelegation(NS2, createStagedWithWriteDelegationRequest); + .createTableStagedWithWriteDelegation( + NS2, + createStagedWithWriteDelegationRequest, + Set.of(AccessDelegationMode.VENDED_CREDENTIALS)); }); } @@ -892,7 +904,10 @@ public void testLoadTableWithReadAccessDelegationSufficientPrivileges() { PolarisPrivilege.TABLE_READ_DATA, PolarisPrivilege.TABLE_WRITE_DATA, PolarisPrivilege.CATALOG_MANAGE_CONTENT), - () -> newWrapper().loadTableWithAccessDelegation(TABLE_NS1A_2, "all"), + () -> + newWrapper() + .loadTableWithAccessDelegation( + TABLE_NS1A_2, Set.of(AccessDelegationMode.VENDED_CREDENTIALS)), null /* cleanupAction */); } @@ -908,7 +923,10 @@ public void testLoadTableWithReadAccessDelegationInsufficientPermissions() { PolarisPrivilege.TABLE_CREATE, PolarisPrivilege.TABLE_LIST, PolarisPrivilege.TABLE_DROP), - () -> newWrapper().loadTableWithAccessDelegation(TABLE_NS1A_2, "all")); + () -> + newWrapper() + .loadTableWithAccessDelegation( + TABLE_NS1A_2, Set.of(AccessDelegationMode.VENDED_CREDENTIALS))); } @Test @@ -921,7 +939,10 @@ public void testLoadTableWithWriteAccessDelegationSufficientPrivileges() { PolarisPrivilege.TABLE_READ_DATA, PolarisPrivilege.TABLE_WRITE_DATA, PolarisPrivilege.CATALOG_MANAGE_CONTENT), - () -> newWrapper().loadTableWithAccessDelegation(TABLE_NS1A_2, "all"), + () -> + newWrapper() + .loadTableWithAccessDelegation( + TABLE_NS1A_2, Set.of(AccessDelegationMode.VENDED_CREDENTIALS)), null /* cleanupAction */); } @@ -937,7 +958,10 @@ public void testLoadTableWithWriteAccessDelegationInsufficientPermissions() { PolarisPrivilege.TABLE_CREATE, PolarisPrivilege.TABLE_LIST, PolarisPrivilege.TABLE_DROP), - () -> newWrapper().loadTableWithAccessDelegation(TABLE_NS1A_2, "all")); + () -> + newWrapper() + .loadTableWithAccessDelegation( + TABLE_NS1A_2, Set.of(AccessDelegationMode.VENDED_CREDENTIALS))); } @Test @@ -950,7 +974,9 @@ public void testLoadTableWithReadAccessDelegationIfStaleSufficientPrivileges() { () -> newWrapper() .loadTableWithAccessDelegationIfStale( - TABLE_NS1A_2, IfNoneMatch.fromHeader("W/\"0:0\""), "all"), + TABLE_NS1A_2, + IfNoneMatch.fromHeader("W/\"0:0\""), + Set.of(AccessDelegationMode.VENDED_CREDENTIALS)), null /* cleanupAction */); } @@ -969,7 +995,45 @@ public void testLoadTableWithReadAccessDelegationIfStaleInsufficientPermissions( () -> newWrapper() .loadTableWithAccessDelegationIfStale( - TABLE_NS1A_2, IfNoneMatch.fromHeader("W/\"0:0\""), "all")); + TABLE_NS1A_2, + IfNoneMatch.fromHeader("W/\"0:0\""), + Set.of(AccessDelegationMode.VENDED_CREDENTIALS))); + } + + @Test + public void testLoadTableWithReadRemoteSigningIfStaleSufficientPrivileges() { + doTestSufficientPrivileges( + List.of( + PolarisPrivilege.TABLE_READ_DATA, + PolarisPrivilege.TABLE_WRITE_DATA, + PolarisPrivilege.CATALOG_MANAGE_CONTENT), + () -> + newWrapper() + .loadTableWithAccessDelegationIfStale( + TABLE_NS1A_2, + IfNoneMatch.fromHeader("W/\"0:0\""), + Set.of(AccessDelegationMode.REMOTE_SIGNING)), + null /* cleanupAction */); + } + + @Test + public void testLoadTableWithReadRemoteSigningIfStaleInsufficientPermissions() { + doTestInsufficientPrivileges( + List.of( + PolarisPrivilege.NAMESPACE_FULL_METADATA, + PolarisPrivilege.VIEW_FULL_METADATA, + PolarisPrivilege.TABLE_FULL_METADATA, + PolarisPrivilege.TABLE_READ_PROPERTIES, + PolarisPrivilege.TABLE_WRITE_PROPERTIES, + PolarisPrivilege.TABLE_CREATE, + PolarisPrivilege.TABLE_LIST, + PolarisPrivilege.TABLE_DROP), + () -> + newWrapper() + .loadTableWithAccessDelegationIfStale( + TABLE_NS1A_2, + IfNoneMatch.fromHeader("W/\"0:0\""), + Set.of(AccessDelegationMode.REMOTE_SIGNING))); } @Test @@ -985,7 +1049,9 @@ public void testLoadTableWithWriteAccessDelegationIfStaleSufficientPrivileges() () -> newWrapper() .loadTableWithAccessDelegationIfStale( - TABLE_NS1A_2, IfNoneMatch.fromHeader("W/\"0:0\""), "all"), + TABLE_NS1A_2, + IfNoneMatch.fromHeader("W/\"0:0\""), + Set.of(AccessDelegationMode.VENDED_CREDENTIALS)), null /* cleanupAction */); } @@ -1004,7 +1070,9 @@ public void testLoadTableWithWriteAccessDelegationIfStaleInsufficientPermissions () -> newWrapper() .loadTableWithAccessDelegationIfStale( - TABLE_NS1A_2, IfNoneMatch.fromHeader("W/\"0:0\""), "all")); + TABLE_NS1A_2, + IfNoneMatch.fromHeader("W/\"0:0\""), + Set.of(AccessDelegationMode.VENDED_CREDENTIALS))); } @Test @@ -1766,8 +1834,12 @@ public void testSendNotificationSufficientPrivileges() { resolverFactory, managerFactory, Mockito.mock(), - new DefaultFileIOFactory(storageCredentialCache, managerFactory), - polarisEventListener) { + new DefaultFileIOFactory( + storageConfiguration, storageCredentialCache, managerFactory, clock), + polarisEventListener, + storageIntegrationProvider, + prefixParser, + Mockito.mock()) { @Override public Catalog createCallContextCatalog( CallContext context, diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/io/FileIOFactoryTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/io/FileIOFactoryTest.java index fd6e79f123..a7defd3afa 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/io/FileIOFactoryTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/io/FileIOFactoryTest.java @@ -20,10 +20,12 @@ import static org.apache.iceberg.types.Types.NestedField.required; import static org.mockito.ArgumentMatchers.isA; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import com.google.common.collect.ImmutableMap; import jakarta.annotation.Nonnull; +import jakarta.ws.rs.core.UriInfo; import java.lang.reflect.Method; import java.util.List; import java.util.Map; @@ -45,7 +47,10 @@ import org.apache.polaris.core.entity.PolarisBaseEntity; import org.apache.polaris.core.entity.TaskEntity; import org.apache.polaris.core.persistence.pagination.PageToken; +import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider; import org.apache.polaris.service.TestServices; +import org.apache.polaris.service.catalog.CatalogPrefixParser; +import org.apache.polaris.service.catalog.DefaultCatalogPrefixParser; import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView; import org.apache.polaris.service.catalog.iceberg.IcebergCatalog; import org.apache.polaris.service.task.TaskFileIOSupplier; @@ -102,9 +107,10 @@ public void before(TestInfo testInfo) { // Spy FileIOFactory and check if the credentials are passed to the FileIO TestServices.FileIOFactorySupplier fileIOFactorySupplier = - (storageCredentialCache, metaStoreManagerFactory) -> + (storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock) -> Mockito.spy( - new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory) { + new DefaultFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock) { @Override FileIO loadFileIOInternal( @Nonnull String ioImplClassName, @Nonnull Map properties) { @@ -223,6 +229,12 @@ IcebergCatalog createCatalog(TestServices services, String scheme) { services.resolutionManifestFactory(), services.securityContext(), CATALOG_NAME); + + PolarisStorageIntegrationProvider storageIntegrationProvider = + mock(PolarisStorageIntegrationProvider.class); + CatalogPrefixParser prefixParser = new DefaultCatalogPrefixParser(); + UriInfo uriInfo = mock(UriInfo.class); + IcebergCatalog polarisCatalog = new IcebergCatalog( services.storageCredentialCache(), @@ -233,7 +245,10 @@ IcebergCatalog createCatalog(TestServices services, String scheme) { services.securityContext(), services.taskExecutor(), services.fileIOFactory(), - services.polarisEventListener()); + services.polarisEventListener(), + storageIntegrationProvider, + prefixParser, + uriInfo); polarisCatalog.initialize( CATALOG_NAME, ImmutableMap.of( diff --git a/runtime/service/src/test/java/org/apache/polaris/service/it/S3RemoteSigningMinIOIntegrationTest.java b/runtime/service/src/test/java/org/apache/polaris/service/it/S3RemoteSigningMinIOIntegrationTest.java new file mode 100644 index 0000000000..1ed641378d --- /dev/null +++ b/runtime/service/src/test/java/org/apache/polaris/service/it/S3RemoteSigningMinIOIntegrationTest.java @@ -0,0 +1,93 @@ +/* + * 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.it; + +import com.google.common.collect.ImmutableMap; +import io.quarkus.test.junit.QuarkusTest; +import io.quarkus.test.junit.QuarkusTestProfile; +import io.quarkus.test.junit.TestProfile; +import java.net.URI; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.apache.polaris.core.storage.StorageAccessProperty; +import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension; +import org.apache.polaris.service.it.test.PolarisS3RemoteSigningIntegrationTest; +import org.apache.polaris.test.minio.Minio; +import org.apache.polaris.test.minio.MinioAccess; +import org.apache.polaris.test.minio.MinioExtension; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.extension.ExtendWith; + +@QuarkusTest +@TestProfile(S3RemoteSigningMinIOIntegrationTest.Profile.class) +@ExtendWith(MinioExtension.class) +@ExtendWith(PolarisIntegrationTestExtension.class) +public class S3RemoteSigningMinIOIntegrationTest extends PolarisS3RemoteSigningIntegrationTest { + + private static final String BUCKET_URI_PREFIX = "/minio-test"; + private static final String MINIO_ACCESS_KEY = "test-ak-123"; + private static final String MINIO_SECRET_KEY = "test-sk-123"; + + public static class Profile implements QuarkusTestProfile { + + @Override + public Map getConfigOverrides() { + return ImmutableMap.builder() + // Polaris will use these keys for remote signing + .put("polaris.storage.aws.access-key", MINIO_ACCESS_KEY) + .put("polaris.storage.aws.secret-key", MINIO_SECRET_KEY) + .put("polaris.features.\"REMOTE_SIGNING_ENABLED\"", "true") + .put("polaris.features.\"SUPPORTED_CATALOG_STORAGE_TYPES\"", "[\"S3\"]") + // Required for MinIO because Polaris needs subscoped credentials + // in order to get the right endpoint URL to use when writing metadata files. + .put("polaris.features.\"SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION\"", "false") + .build(); + } + } + + private static URI storageBase; + private static String endpoint; + + @BeforeAll + static void setup( + @Minio(accessKey = MINIO_ACCESS_KEY, secretKey = MINIO_SECRET_KEY) MinioAccess minioAccess) { + storageBase = minioAccess.s3BucketUri(BUCKET_URI_PREFIX); + endpoint = minioAccess.s3endpoint(); + } + + @Override + protected List allowedLocations() { + return List.of(storageBase.toString()); + } + + @Override + protected Optional endpoint() { + return Optional.of(endpoint); + } + + @Override + protected ImmutableMap.Builder clientFileIOProperties() { + return super.clientFileIOProperties() + // Grant direct access to the MinIO bucket; this FileIO instance does not + // use access delegation. + .put(StorageAccessProperty.AWS_KEY_ID.getPropertyName(), MINIO_ACCESS_KEY) + .put(StorageAccessProperty.AWS_SECRET_KEY.getPropertyName(), MINIO_SECRET_KEY); + } +} diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/StorageConfigurationTest.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/StorageConfigurationTest.java index 916b1912a1..03ec4f9744 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/storage/StorageConfigurationTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/StorageConfigurationTest.java @@ -192,7 +192,7 @@ public void testSingletonStsClientWithStaticCredentials() { staticMock.when(StsClient::builder).thenReturn(mockBuilder); StorageConfiguration config = configWithAwsCredentialsAndGcpToken(); - Supplier supplier = config.stsClientSupplier(true); + Supplier supplier = config.stsClientSupplier(); StsClient client1 = supplier.get(); StsClient client2 = supplier.get(); @@ -209,7 +209,7 @@ public void testSingletonStsClientWithStaticCredentials() { @Test public void testStaticStsCredentials() { StorageConfiguration config = configWithAwsCredentialsAndGcpToken(); - AwsCredentialsProvider credentialsProvider = config.stsCredentials(); + AwsCredentialsProvider credentialsProvider = config.awsSystemCredentials(); assertThat(credentialsProvider).isInstanceOf(StaticCredentialsProvider.class); assertThat(credentialsProvider.resolveCredentials().accessKeyId()).isEqualTo(TEST_ACCESS_KEY); assertThat(credentialsProvider.resolveCredentials().secretAccessKey()) diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogHandlerAuthzTest.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogHandlerAuthzTest.java new file mode 100644 index 0000000000..1eafe9f8ea --- /dev/null +++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/s3/sign/S3RemoteSigningCatalogHandlerAuthzTest.java @@ -0,0 +1,120 @@ +/* + * 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.storage.s3.sign; + +import static org.mockito.ArgumentMatchers.any; + +import io.quarkus.test.junit.QuarkusTest; +import io.quarkus.test.junit.TestProfile; +import java.net.URI; +import java.util.List; +import java.util.Set; +import org.apache.polaris.core.auth.PolarisPrincipal; +import org.apache.polaris.core.entity.PolarisPrivilege; +import org.apache.polaris.service.admin.PolarisAuthzTestBase; +import org.apache.polaris.service.s3.sign.model.ImmutablePolarisS3SignRequest; +import org.apache.polaris.service.s3.sign.model.ImmutablePolarisS3SignResponse; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +@QuarkusTest +@TestProfile(PolarisAuthzTestBase.Profile.class) +@SuppressWarnings("resource") +public class S3RemoteSigningCatalogHandlerAuthzTest extends PolarisAuthzTestBase { + + private static final ImmutablePolarisS3SignRequest READ_REQUEST = + ImmutablePolarisS3SignRequest.builder() + .method("GET") + .uri(URI.create("https://example-bucket.s3.amazonaws.com/some-object")) + .region("us-west-2") + .build(); + + private static final ImmutablePolarisS3SignRequest WRITE_REQUEST = + ImmutablePolarisS3SignRequest.builder() + .method("PUT") + .uri(URI.create("https://example-bucket.s3.amazonaws.com/some-object")) + .region("us-west-2") + .build(); + + @Test + public void testReadRequestInsufficientPermissions() { + doTestInsufficientPrivileges( + List.of(PolarisPrivilege.TABLE_REMOTE_SIGN, PolarisPrivilege.TABLE_READ_DATA), + () -> newHandler().signS3Request(READ_REQUEST, TABLE_NS1_1)); + } + + @Test + public void testWriteRequestInsufficientPermissions() { + doTestInsufficientPrivileges( + List.of(PolarisPrivilege.TABLE_REMOTE_SIGN, PolarisPrivilege.TABLE_WRITE_DATA), + () -> newHandler().signS3Request(WRITE_REQUEST, TABLE_NS1_1)); + } + + @Test + public void testReadRequestSufficientPermissions() { + doTestSufficientPrivilegeSets( + List.of(Set.of(PolarisPrivilege.TABLE_READ_DATA, PolarisPrivilege.TABLE_REMOTE_SIGN)), + () -> newHandler().signS3Request(READ_REQUEST, TABLE_NS1_1), + () -> {}, + PRINCIPAL_NAME, + (privilege) -> + adminService.grantPrivilegeOnCatalogToRole(CATALOG_NAME, CATALOG_ROLE1, privilege), + (privilege) -> + adminService.revokePrivilegeOnCatalogFromRole(CATALOG_NAME, CATALOG_ROLE1, privilege)); + } + + @Test + public void testWriteRequestSufficientPermissions() { + doTestSufficientPrivilegeSets( + List.of(Set.of(PolarisPrivilege.TABLE_WRITE_DATA, PolarisPrivilege.TABLE_REMOTE_SIGN)), + () -> newHandler().signS3Request(WRITE_REQUEST, TABLE_NS1_1), + () -> {}, + PRINCIPAL_NAME, + (privilege) -> + adminService.grantPrivilegeOnCatalogToRole(CATALOG_NAME, CATALOG_ROLE1, privilege), + (privilege) -> + adminService.revokePrivilegeOnCatalogFromRole(CATALOG_NAME, CATALOG_ROLE1, privilege)); + } + + private S3RemoteSigningCatalogHandler newHandler() { + PolarisPrincipal principal = PolarisPrincipal.of(principalEntity, Set.of()); + S3RequestSigner s3signer = Mockito.mock(S3RequestSigner.class); + Mockito.when(s3signer.signRequest(any())) + .thenReturn(ImmutablePolarisS3SignResponse.builder().uri(URI.create("irrelevant")).build()); + return new S3RemoteSigningCatalogHandler( + callContext, + resolutionManifestFactory, + securityContext(principal), + CATALOG_NAME, + polarisAuthorizer, + s3signer); + } + + private void doTestInsufficientPrivileges( + List insufficientPrivileges, Runnable action) { + doTestInsufficientPrivileges( + insufficientPrivileges, + PRINCIPAL_NAME, + action, + (privilege) -> + adminService.grantPrivilegeOnCatalogToRole(CATALOG_NAME, CATALOG_ROLE1, privilege), + (privilege) -> + adminService.revokePrivilegeOnCatalogFromRole(CATALOG_NAME, CATALOG_ROLE1, privilege)); + } +} diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/s3/sign/S3RequestSignerImplTest.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/s3/sign/S3RequestSignerImplTest.java new file mode 100644 index 0000000000..9f55d6faac --- /dev/null +++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/s3/sign/S3RequestSignerImplTest.java @@ -0,0 +1,221 @@ +/* + * 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.storage.s3.sign; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.when; + +import java.net.URI; +import java.util.List; +import java.util.Map; +import org.apache.iceberg.aws.s3.signer.ImmutableS3SignRequest; +import org.apache.iceberg.aws.s3.signer.S3SignRequest; +import org.apache.iceberg.aws.s3.signer.S3SignResponse; +import org.apache.polaris.service.storage.StorageConfiguration; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; + +@ExtendWith(MockitoExtension.class) +class S3RequestSignerImplTest { + + @Mock private StorageConfiguration storageConfiguration; + @Mock private AwsCredentialsProvider awsCredentialsProvider; + + @InjectMocks private S3RequestSignerImpl s3RequestSigner; + + private static final String TEST_ACCESS_KEY = "test-access-key"; + private static final String TEST_SECRET_KEY = "test-secret-key"; + private static final String TEST_REGION = "us-west-2"; + private static final String TEST_HOST = "test-bucket.s3.us-west-2.amazonaws.com"; + private static final URI TEST_URI = URI.create("https://" + TEST_HOST + "/test-path"); + + @BeforeEach + void setUp() { + AwsCredentials credentials = AwsBasicCredentials.create(TEST_ACCESS_KEY, TEST_SECRET_KEY); + when(storageConfiguration.awsSystemCredentials()).thenReturn(awsCredentialsProvider); + when(awsCredentialsProvider.resolveCredentials()).thenReturn(credentials); + } + + @Test + void testGET() { + // Given + S3SignRequest request = + ImmutableS3SignRequest.builder() + .region(TEST_REGION) + .method("GET") + .uri(TEST_URI) + .headers(Map.of("Host", List.of(TEST_HOST))) + .properties(Map.of()) + .build(); + + // When + S3SignResponse response = s3RequestSigner.signRequest(request); + + // Then + assertThat(response).isNotNull(); + assertThat(response.uri()).isEqualTo(TEST_URI); + assertHeaders(response); + } + + @Test + void testPUT() { + // Given + String requestBody = "{\"test\": \"data\"}"; + S3SignRequest request = + ImmutableS3SignRequest.builder() + .region(TEST_REGION) + .method("PUT") + .uri(TEST_URI) + .headers(Map.of("Host", List.of(TEST_HOST))) + .properties(Map.of()) + .body(requestBody) + .build(); + + // When + S3SignResponse response = s3RequestSigner.signRequest(request); + + // Then + assertThat(response).isNotNull(); + assertThat(response.uri()).isEqualTo(TEST_URI); + assertHeaders(response); + } + + @Test + void testPOST() { + // Given + String requestBody = "{\"test\": \"data\"}"; + S3SignRequest request = + ImmutableS3SignRequest.builder() + .region(TEST_REGION) + .method("POST") + .uri(TEST_URI) + .headers(Map.of("Host", List.of(TEST_HOST))) + .properties(Map.of()) + .body(requestBody) + .build(); + + // When + S3SignResponse response = s3RequestSigner.signRequest(request); + + // Then + assertThat(response).isNotNull(); + assertThat(response.uri()).isEqualTo(TEST_URI); + assertHeaders(response); + } + + @Test + void testDELETE() { + // Given + String requestBody = "{\"test\": \"data\"}"; + S3SignRequest request = + ImmutableS3SignRequest.builder() + .region(TEST_REGION) + .method("DELETE") + .uri(TEST_URI) + .headers(Map.of("Host", List.of(TEST_HOST))) + .properties(Map.of()) + .body(requestBody) + .build(); + + // When + S3SignResponse response = s3RequestSigner.signRequest(request); + + // Then + assertThat(response).isNotNull(); + assertThat(response.uri()).isEqualTo(TEST_URI); + assertHeaders(response); + } + + @Test + void testQueryParameters() { + // Given + URI uriWithQuery = URI.create(TEST_URI + "?prefix=test&max-keys=100"); + S3SignRequest request = + ImmutableS3SignRequest.builder() + .region(TEST_REGION) + .method("GET") + .uri(uriWithQuery) + .headers(Map.of("Host", List.of(TEST_HOST))) + .properties(Map.of()) + .build(); + + // When + S3SignResponse response = s3RequestSigner.signRequest(request); + + // Then + assertThat(response).isNotNull(); + assertThat(response.uri().getHost()).isEqualTo(uriWithQuery.getHost()); + assertThat(response.uri().getPath()).isEqualTo(uriWithQuery.getPath()); + assertThat(response.uri().getQuery()).contains("prefix=test"); + assertThat(response.uri().getQuery()).contains("max-keys=100"); + assertHeaders(response); + } + + @Test + void testHeaders() { + // Given + S3SignRequest request = + ImmutableS3SignRequest.builder() + .region(TEST_REGION) + .method("GET") + .uri(TEST_URI) + .headers( + Map.of( + "Host", List.of(TEST_HOST), + "x-amz-content-sha256", List.of("UNSIGNED-PAYLOAD"), + "Content-Type", List.of("application/json"), + "User-Agent", List.of("test-client/1.0"))) + .properties(Map.of()) + .build(); + + // When + S3SignResponse response = s3RequestSigner.signRequest(request); + + // Then + assertThat(response).isNotNull(); + assertThat(response.uri()).isEqualTo(TEST_URI); + assertHeaders(response); + assertThat(response.headers().get("x-amz-content-sha256").getFirst()) + .isEqualTo("UNSIGNED-PAYLOAD"); + assertThat(response.headers().get("Content-Type").getFirst()).isEqualTo("application/json"); + assertThat(response.headers().get("User-Agent").getFirst()).isEqualTo("test-client/1.0"); + } + + private static void assertHeaders(S3SignResponse response) { + assertThat(response.headers()).isNotEmpty(); + assertThat(response.headers()).containsKey("X-Amz-Date"); + assertThat(response.headers()).containsKey("x-amz-content-sha256"); + assertThat(response.headers().get("x-amz-content-sha256").getFirst()) + .isEqualTo("UNSIGNED-PAYLOAD"); // Fixed for S3 + assertThat(response.headers()).containsKey("Authorization"); + assertThat(response.headers().get("Authorization")).hasSize(1); + String authHeader = response.headers().get("Authorization").getFirst(); + assertThat(authHeader).startsWith("AWS4-HMAC-SHA256 Credential=" + TEST_ACCESS_KEY); + assertThat(authHeader).contains("SignedHeaders="); + assertThat(authHeader).contains("Signature="); + } +} diff --git a/runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java b/runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java index 2cb9a483de..64fbb45032 100644 --- a/runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java +++ b/runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java @@ -31,7 +31,6 @@ import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.function.BiFunction; import org.apache.polaris.core.PolarisCallContext; import org.apache.polaris.core.PolarisDefaultDiagServiceImpl; import org.apache.polaris.core.PolarisDiagnostics; @@ -58,6 +57,7 @@ import org.apache.polaris.core.storage.cache.StorageCredentialCacheConfig; import org.apache.polaris.service.admin.PolarisServiceImpl; import org.apache.polaris.service.admin.api.PolarisCatalogsApi; +import org.apache.polaris.service.catalog.CatalogPrefixParser; import org.apache.polaris.service.catalog.DefaultCatalogPrefixParser; import org.apache.polaris.service.catalog.api.IcebergRestCatalogApi; import org.apache.polaris.service.catalog.api.IcebergRestConfigurationApi; @@ -73,6 +73,7 @@ import org.apache.polaris.service.persistence.InMemoryPolarisMetaStoreManagerFactory; import org.apache.polaris.service.secrets.UnsafeInMemorySecretsManagerFactory; import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl; +import org.apache.polaris.service.storage.StorageConfiguration; import org.apache.polaris.service.task.TaskExecutor; import org.mockito.Mockito; import software.amazon.awssdk.services.sts.StsClient; @@ -100,8 +101,13 @@ public record TestServices( private static final String GCP_ACCESS_TOKEN = "abc"; @FunctionalInterface - public interface FileIOFactorySupplier - extends BiFunction {} + public interface FileIOFactorySupplier { + FileIOFactory supplyFileIOFactory( + StorageConfiguration storageConfiguration, + StorageCredentialCache storageCredentialCache, + MetaStoreManagerFactory metaStoreManagerFactory, + Clock clock); + } private static class MockedConfigurationStore implements PolarisConfigurationStore { private final Map defaults; @@ -198,12 +204,18 @@ public TestServices build() { UserSecretsManager userSecretsManager = userSecretsManagerFactory.getOrCreateUserSecretsManager(realmContext); + StorageConfiguration storageConfiguration = + Mockito.mock(StorageConfiguration.class, Mockito.RETURNS_DEFAULTS); + FileIOFactory fileIOFactory = - fileIOFactorySupplier.apply(storageCredentialCache, metaStoreManagerFactory); + fileIOFactorySupplier.supplyFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock); TaskExecutor taskExecutor = Mockito.mock(TaskExecutor.class); PolarisEventListener polarisEventListener = new TestPolarisEventListener(); + CatalogPrefixParser prefixParser = new DefaultCatalogPrefixParser(); + CallContextCatalogFactory callContextFactory = new PolarisCallContextCatalogFactory( storageCredentialCache, @@ -211,7 +223,10 @@ public TestServices build() { metaStoreManagerFactory, taskExecutor, fileIOFactory, - polarisEventListener); + polarisEventListener, + storageIntegrationProvider, + prefixParser, + Mockito.mock()); ReservedProperties reservedProperties = ReservedProperties.NONE; diff --git a/runtime/service/src/testFixtures/java/org/apache/polaris/service/catalog/io/MeasuredFileIOFactory.java b/runtime/service/src/testFixtures/java/org/apache/polaris/service/catalog/io/MeasuredFileIOFactory.java index c4bf40ca92..9333ee5159 100644 --- a/runtime/service/src/testFixtures/java/org/apache/polaris/service/catalog/io/MeasuredFileIOFactory.java +++ b/runtime/service/src/testFixtures/java/org/apache/polaris/service/catalog/io/MeasuredFileIOFactory.java @@ -21,6 +21,7 @@ import jakarta.annotation.Nonnull; import jakarta.enterprise.inject.Vetoed; import jakarta.inject.Inject; +import java.time.Clock; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -34,6 +35,7 @@ import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper; import org.apache.polaris.core.storage.PolarisStorageActions; import org.apache.polaris.core.storage.cache.StorageCredentialCache; +import org.apache.polaris.service.storage.StorageConfiguration; /** * A FileIOFactory that measures the number of bytes read, files written, and files deleted. It can @@ -53,10 +55,13 @@ public class MeasuredFileIOFactory implements FileIOFactory { @Inject public MeasuredFileIOFactory( + StorageConfiguration storageConfiguration, StorageCredentialCache storageCredentialCache, - MetaStoreManagerFactory metaStoreManagerFactory) { + MetaStoreManagerFactory metaStoreManagerFactory, + Clock clock) { defaultFileIOFactory = - new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory); + new DefaultFileIOFactory( + storageConfiguration, storageCredentialCache, metaStoreManagerFactory, clock); } @Override diff --git a/site/content/in-dev/unreleased/access-control.md b/site/content/in-dev/unreleased/access-control.md index e8ed4b5e56..13d8fb4cc9 100644 --- a/site/content/in-dev/unreleased/access-control.md +++ b/site/content/in-dev/unreleased/access-control.md @@ -103,18 +103,20 @@ To grant the full set of privileges (drop, list, read, write, etc.) on an object ### Table privileges -| Privilege | Description | -| --------- | ----------- | -| TABLE_CREATE | Enables registering a table with the catalog. | -| TABLE_DROP | Enables dropping a table from the catalog. | -| TABLE_LIST | Enables listing any table in the catalog. | -| TABLE_READ_PROPERTIES | Enables reading properties of the table. | -| TABLE_WRITE_PROPERTIES | Enables configuring properties for the table. | -| TABLE_READ_DATA | Enables reading data from the table by receiving short-lived read-only storage credentials from the catalog. | -| TABLE_WRITE_DATA | Enables writing data to the table by receiving short-lived read+write storage credentials from the catalog. | -| TABLE_FULL_METADATA | Grants all table privileges, except TABLE_READ_DATA and TABLE_WRITE_DATA, which need to be granted individually. | -| TABLE_ATTACH_POLICY | Enables attaching policy to a table. | -| TABLE_DETACH_POLICY | Enables detaching policy from a table. | +| Privilege | Description | +|------------------------|------------------------------------------------------------------------------------------------------------------| +| TABLE_CREATE | Enables registering a table with the catalog. | +| TABLE_DROP | Enables dropping a table from the catalog. | +| TABLE_LIST | Enables listing any table in the catalog. | +| TABLE_READ_PROPERTIES | Enables reading properties of the table. | +| TABLE_WRITE_PROPERTIES | Enables configuring properties for the table. | +| TABLE_READ_DATA | Enables reading data from the table. | +| TABLE_WRITE_DATA | Enables writing data to the table. | +| TABLE_FULL_METADATA | Grants all table privileges, except TABLE_READ_DATA and TABLE_WRITE_DATA, which need to be granted individually. | +| TABLE_ATTACH_POLICY | Enables attaching policy to a table. | +| TABLE_DETACH_POLICY | Enables detaching policy from a table. | +| TABLE_REMOTE_SIGN | Enables remote signing for a table. TABLE_READ_DATA and TABLE_WRITE_DATA must also be granted individually. | + ### View privileges diff --git a/spec/README.md b/spec/README.md index e168dbeda5..6e1bf2c6f0 100644 --- a/spec/README.md +++ b/spec/README.md @@ -17,14 +17,40 @@ under the License. --> -# Polaris API Specifications - -Polaris provides two sets of OpenAPI specifications: -- `polaris-management-service.yml` - Defines the management APIs for using Polaris to create and manage Iceberg catalogs and their principals -- `polaris-catalog-service.yaml` - Defines the specification for the Polaris Catalog API, which encompasses both the Iceberg REST Catalog API - and Polaris-native API. - - `polaris-catalog-apis` - Contains the specification for Polaris-specific Catalog APIs - - `iceberg-rest-catalog-open-api.yaml` - Contains the specification for Iceberg Rest Catalog API +# Apache Polaris API Specifications + +Apache Polaris provides the following OpenAPI specifications: + +- [polaris-management-service.yml](polaris-management-service.yml) - Defines the management APIs for using Apache + Polaris to create and manage Apache Iceberg catalogs and their principals + +- [polaris-catalog-service.yaml](polaris-catalog-service.yaml) - Defines the specification for the Apache Polaris + Catalog API, which encompasses both the Apache Iceberg REST Catalog API and Apache Polaris-native APIs: + + - [iceberg-rest-catalog-open-api.yaml](iceberg-rest-catalog-open-api.yaml) - Contains the specification for + Apache Iceberg Rest Catalog API. + + - [polaris-catalog-apis](polaris-catalog-apis) - This folder contains the specifications for Apache + Polaris-specific Catalog APIs: + + - [generic-tables-api.yaml](polaris-catalog-apis/generic-tables-api.yaml) - Contains the specification for + the Generic Tables API + + - [notifications-api.yaml](polaris-catalog-apis/notifications-api.yaml) - Contains the specification for + the Notifications API + + - [oauth-tokens-api.yaml](polaris-catalog-apis/oauth-tokens-api.yaml) - Contains the specification for the + internal OAuth Token endpoint, extracted from the Apache Iceberg REST Catalog API. + + - [policy-apis.yaml](polaris-catalog-apis/policy-apis.yaml) - Contains the specification for the Policy APIs. + +- [s3-sign](s3-sign) - This folder contains the specifications for S3 remote signing: + + - [iceberg-s3-signer-open-api.yaml](s3-sign/iceberg-s3-signer-open-api.yaml) - Contains the specification of the + original Apache Iceberg S3 Signer API. + + - [polaris-s3-sign-service.yaml](s3-sign/polaris-s3-sign-service.yaml) - Contains the Apache Polaris-specific + S3 Signer API. ## Generated Specification Files The specification files in the generated folder are automatically created using OpenAPI bundling tools such as diff --git a/spec/s3-sign/iceberg-s3-signer-open-api.yaml b/spec/s3-sign/iceberg-s3-signer-open-api.yaml new file mode 100644 index 0000000000..bd292f239b --- /dev/null +++ b/spec/s3-sign/iceberg-s3-signer-open-api.yaml @@ -0,0 +1,154 @@ +# +# 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. +# + +# CODE_COPIED_TO_POLARIS + +# Apache Iceberg Version: 1.9.2 + +--- +openapi: 3.0.3 +info: + title: Apache Iceberg S3 Signer API + license: + name: Apache 2.0 + url: https://www.apache.org/licenses/LICENSE-2.0.html + version: 0.0.1 + description: + Defines the specification for the S3 Signer API. +servers: + - url: "{scheme}://{host}/{basePath}" + description: Server URL when the port can be inferred from the scheme + variables: + scheme: + description: The scheme of the URI, either http or https. + default: https + host: + description: The host address for the specified server + default: localhost + basePath: + description: Optional prefix to be prepended to all routes + default: "" + - url: "{scheme}://{host}:{port}/{basePath}" + description: Generic base server URL, with all parts configurable + variables: + scheme: + description: The scheme of the URI, either http or https. + default: https + host: + description: The host address for the specified server + default: localhost + port: + description: The port used when addressing the host + default: "443" + basePath: + description: Optional prefix to be appended to all routes + default: "" + +paths: + + /v1/aws/s3/sign: + + post: + tags: + - S3 Signer API + summary: Remotely signs S3 requests + operationId: signS3Request + requestBody: + description: The request containing the headers to be signed + content: + application/json: + schema: + $ref: '#/components/schemas/S3SignRequest' + required: true + responses: + 200: + $ref: '#/components/responses/S3SignResponse' + 400: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse' + 401: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse' + 403: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse' + 419: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/AuthenticationTimeoutResponse' + 503: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse' + 5XX: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse' + + ############################## + # Application Schema Objects # + ############################## +components: + schemas: + + S3Headers: + type: object + additionalProperties: + type: array + items: + type: string + + S3SignRequest: + required: + - region + - uri + - method + - headers + properties: + region: + type: string + uri: + type: string + method: + type: string + enum: ["PUT", "GET", "HEAD", "POST", "DELETE", "PATCH", "OPTIONS"] + headers: + $ref: '#/components/schemas/S3Headers' + properties: + type: object + additionalProperties: + type: string + body: + type: string + description: Optional body of the S3 request to send to the signing API. This should only be populated + for S3 requests which do not have the relevant data in the URI itself (e.g. DeleteObjects requests) + + + ############################# + # Reusable Response Objects # + ############################# + responses: + + S3SignResponse: + description: The response containing signed & unsigned headers. The server will also send + a Cache-Control header, indicating whether the response can be cached (Cache-Control = ["private"]) + or not (Cache-Control = ["no-cache"]). + content: + application/json: + schema: + type: object + required: + - uri + - headers + properties: + uri: + type: string + headers: + $ref: '#/components/schemas/S3Headers' diff --git a/spec/s3-sign/polaris-s3-sign-service.yaml b/spec/s3-sign/polaris-s3-sign-service.yaml new file mode 100644 index 0000000000..7e37b44117 --- /dev/null +++ b/spec/s3-sign/polaris-s3-sign-service.yaml @@ -0,0 +1,108 @@ +# +# 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. +# + +# Polaris-specific S3 Signer API that includes path parameters required for Polaris, such as +# the prefix, namespace, and table. + +openapi: 3.0.3 +info: + title: Apache Polaris S3 Signer API + license: + name: Apache 2.0 + url: https://www.apache.org/licenses/LICENSE-2.0.html + version: 0.0.1 + description: + Defines the specification for the S3 Signer API integrated with Polaris catalog structure. + +# The server configuration is sourced from iceberg-s3-signer-open-api.yaml +servers: + - url: "{scheme}://{host}/{basePath}" + description: Server URL when the port can be inferred from the scheme + variables: + scheme: + description: The scheme of the URI, either http or https. + default: https + host: + description: The host address for the specified server + default: localhost + basePath: + description: Optional prefix to be appended to all routes + default: "" + - url: "{scheme}://{host}:{port}/{basePath}" + description: Generic base server URL, with all parts configurable + variables: + scheme: + description: The scheme of the URI, either http or https. + default: https + host: + description: The host address for the specified server + default: localhost + port: + description: The port used when addressing the host + default: "443" + +# All routes are currently configured using an Authorization header. +security: + - OAuth2: [] + +paths: + + /v1/{prefix}/namespaces/{namespace}/tables/{table}: + parameters: + - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/prefix' + - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/namespace' + - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/table' + + post: + tags: + - S3 Signer API + summary: Remotely signs S3 requests + operationId: signS3Request + requestBody: + description: The request containing the headers to be signed + content: + application/json: + schema: + $ref: './iceberg-s3-signer-open-api.yaml#/components/schemas/S3SignRequest' + required: true + responses: + 200: + $ref: './iceberg-s3-signer-open-api.yaml#/components/responses/S3SignResponse' + 400: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse' + 401: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse' + 403: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse' + 419: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/AuthenticationTimeoutResponse' + 503: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse' + 5XX: + $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse' + +components: + securitySchemes: + OAuth2: + type: oauth2 + description: Uses OAuth 2 with client credentials flow + flows: + clientCredentials: + tokenUrl: "{scheme}://{host}/api/v1/oauth/tokens" + scopes: {} \ No newline at end of file