You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@polaris.apache.org by sn...@apache.org on 2026/07/22 09:37:45 UTC
(polaris) branch main updated: Add Floci S3 runtime-service integration tests (#5101)
This is an automated email from the ASF dual-hosted git repository.
snazy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
The following commit(s) were added to refs/heads/main by this push:
new b8c45e387b Add Floci S3 runtime-service integration tests (#5101)
b8c45e387b is described below
commit b8c45e387bc783b0bc9951a07eb0c1b40e65f299
Author: Robert Stupp <sn...@snazy.de>
AuthorDate: Wed Jul 22 11:37:40 2026 +0200
Add Floci S3 runtime-service integration tests (#5101)
Add runtime-service integration coverage for the Floci AWS testcontainer so the MinIO-backed REST catalog tests can be migrated with equivalent local coverage.
The new tests cover:
- standard S3 REST catalog table read/write behavior
- S3 special-character and path handling scenarios
- S3 credential-vending behavior backed by Floci STS/IAM
- optional KMS-backed credential-vending coverage
This follows the dev@ mailing-list discussion [Discussion: minio replacement for S3 compatible storage](https://lists.apache.org/thread/c8t4rzd92f5fhg4k9qtyhkv7tzls10yf), where Floci was discussed as a local emulator candidate for S3-compatible storage tests that need AWS service behavior beyond object storage.
The tests use Floci because this coverage depends on AWS-style services such as STS, IAM, and KMS, not only S3 object-store compatibility.
---
.../it/AbstractRestCatalogFlociS3SpecialIT.java | 395 +++++++++++++++++++++
.../service/it/PolarisRestCatalogFlociS3IT.java | 158 +++++++++
.../service/it/RestCatalogFlociS3SpecialIT.java | 285 +++++++++++++++
.../service/it/RestCatalogFlociS3SpecialKmsIT.java | 57 +++
4 files changed, 895 insertions(+)
diff --git a/runtime/service/src/intTest/java/org/apache/polaris/service/it/AbstractRestCatalogFlociS3SpecialIT.java b/runtime/service/src/intTest/java/org/apache/polaris/service/it/AbstractRestCatalogFlociS3SpecialIT.java
new file mode 100644
index 0000000000..e00784015e
--- /dev/null
+++ b/runtime/service/src/intTest/java/org/apache/polaris/service/it/AbstractRestCatalogFlociS3SpecialIT.java
@@ -0,0 +1,395 @@
+/*
+ * 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 static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.iceberg.CatalogProperties.TABLE_DEFAULT_PREFIX;
+import static org.apache.iceberg.aws.AwsClientProperties.REFRESH_CREDENTIALS_ENDPOINT;
+import static org.apache.iceberg.aws.s3.S3FileIOProperties.ENDPOINT;
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.apache.polaris.core.storage.StorageAccessProperty.AWS_KEY_ID;
+import static org.apache.polaris.core.storage.StorageAccessProperty.AWS_SECRET_KEY;
+import static org.apache.polaris.service.catalog.AccessDelegationMode.VENDED_CREDENTIALS;
+import static org.apache.polaris.service.it.env.PolarisClient.polarisClient;
+import static org.apache.polaris.test.commons.MinioRustProfile.ACCESS_KEY;
+import static org.apache.polaris.test.commons.MinioRustProfile.SECRET_KEY;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.google.common.collect.ImmutableMap;
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.rest.auth.OAuth2Properties;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.types.Types;
+import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.CatalogProperties;
+import org.apache.polaris.core.admin.model.PolarisCatalog;
+import org.apache.polaris.core.admin.model.PrincipalWithCredentials;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.service.catalog.AccessDelegationMode;
+import org.apache.polaris.service.it.env.CatalogApi;
+import org.apache.polaris.service.it.env.ClientCredentials;
+import org.apache.polaris.service.it.env.ManagementApi;
+import org.apache.polaris.service.it.env.PolarisApiEndpoints;
+import org.apache.polaris.service.it.env.PolarisClient;
+import org.apache.polaris.test.floci.aws.FlociAwsAccess;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestInfo;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+
+abstract class AbstractRestCatalogFlociS3SpecialIT {
+ protected static final String BUCKET_URI_PREFIX = "/floci-s3-test";
+ protected static final String TEST_REGION = "us-east-1";
+
+ protected static final Schema SCHEMA =
+ new Schema(
+ required(1, "id", Types.IntegerType.get(), "doc"),
+ optional(2, "data", Types.StringType.get()));
+
+ private static String adminToken;
+
+ protected static PolarisApiEndpoints endpoints;
+ protected static PolarisClient client;
+ protected static ManagementApi managementApi;
+ protected static URI storageBase;
+ protected static String endpoint;
+ protected static S3Client s3Client;
+ protected static String roleName;
+ protected static String roleArn;
+
+ protected CatalogApi catalogApi;
+ protected String principalRoleName;
+ protected PrincipalWithCredentials principalCredentials;
+ protected String catalogName;
+
+ protected abstract FlociAwsAccess flociAwsAccess();
+
+ protected boolean kmsUnavailableForStsCatalogs() {
+ return true;
+ }
+
+ @BeforeAll
+ void setup(PolarisApiEndpoints apiEndpoints, ClientCredentials credentials) {
+ var flociAws = flociAwsAccess();
+ endpoints = apiEndpoints;
+ s3Client = flociAws.s3Client();
+ client = polarisClient(endpoints);
+ adminToken = client.obtainToken(credentials);
+ managementApi = client.managementApi(adminToken);
+ storageBase = flociAws.s3BucketUri(BUCKET_URI_PREFIX);
+ endpoint = flociAws.s3endpoint();
+ roleName = "polaris-floci-access-role-" + UUID.randomUUID();
+ roleArn = "arn:aws:iam::" + flociAws.accountId() + ":role/" + roleName;
+ createRoleWithAccess(flociAws);
+ }
+
+ @AfterAll
+ void close() throws Exception {
+ client.close();
+ }
+
+ @BeforeEach
+ public void before(TestInfo testInfo) {
+ var principalName = client.newEntityName("test-user");
+ principalRoleName = client.newEntityName("test-admin");
+ principalCredentials = managementApi.createPrincipalWithRole(principalName, principalRoleName);
+
+ var principalToken = client.obtainToken(principalCredentials);
+ catalogApi = client.catalogApi(principalToken);
+
+ catalogName = client.newEntityName(testInfo.getTestMethod().orElseThrow().getName());
+ }
+
+ @AfterEach
+ public void cleanUp() {
+ client.cleanUp(adminToken);
+ }
+
+ protected RESTCatalog createCatalog(
+ Optional<String> endpoint,
+ Optional<String> stsEndpoint,
+ boolean pathStyleAccess,
+ Optional<AccessDelegationMode> delegationMode,
+ boolean stsEnabled) {
+ return createCatalog(
+ endpoint,
+ stsEndpoint,
+ pathStyleAccess,
+ Optional.empty(),
+ delegationMode,
+ stsEnabled,
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty());
+ }
+
+ protected RESTCatalog createCatalog(
+ Optional<String> endpoint,
+ Optional<String> stsEndpoint,
+ boolean pathStyleAccess,
+ Optional<String> endpointInternal,
+ Optional<AccessDelegationMode> delegationMode,
+ boolean stsEnabled,
+ Optional<String> region,
+ Optional<String> roleArn,
+ Optional<Boolean> kmsUnavailable) {
+ var storageConfigBuilder =
+ AwsStorageConfigInfo.builder()
+ .setStorageType(StorageConfigInfo.StorageTypeEnum.S3)
+ .setPathStyleAccess(pathStyleAccess)
+ .setStsUnavailable(!stsEnabled)
+ .setAllowedLocations(List.of(storageBase.toASCIIString() + "/" + catalogName));
+
+ endpoint.ifPresent(storageConfigBuilder::setEndpoint);
+ stsEndpoint.ifPresent(storageConfigBuilder::setStsEndpoint);
+ endpointInternal.ifPresent(storageConfigBuilder::setEndpointInternal);
+ region.ifPresent(storageConfigBuilder::setRegion);
+ roleArn.ifPresent(storageConfigBuilder::setRoleArn);
+ kmsUnavailable.ifPresent(storageConfigBuilder::setKmsUnavailable);
+
+ var catalogProps = CatalogProperties.builder(storageBase.toASCIIString() + "/" + catalogName);
+ if (!stsEnabled) {
+ catalogProps.addProperty(TABLE_DEFAULT_PREFIX + AWS_KEY_ID.getPropertyName(), ACCESS_KEY);
+ catalogProps.addProperty(TABLE_DEFAULT_PREFIX + AWS_SECRET_KEY.getPropertyName(), SECRET_KEY);
+ }
+ var catalog =
+ PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(catalogName)
+ .setStorageConfigInfo(storageConfigBuilder.build())
+ .setProperties(catalogProps.build())
+ .build();
+
+ managementApi.createCatalog(principalRoleName, catalog);
+
+ var authToken = client.obtainToken(principalCredentials);
+ var restCatalog = new RESTCatalog();
+
+ var propertiesBuilder =
+ ImmutableMap.<String, String>builder()
+ .put(
+ org.apache.iceberg.CatalogProperties.URI, endpoints.catalogApiEndpoint().toString())
+ .put(OAuth2Properties.TOKEN, authToken)
+ .put("warehouse", catalogName)
+ .putAll(endpoints.extraHeaders("header."));
+
+ delegationMode.ifPresent(
+ dm -> propertiesBuilder.put("header.X-Iceberg-Access-Delegation", dm.protocolValue()));
+
+ if (delegationMode.isEmpty()) {
+ propertiesBuilder.put("s3.access-key-id", ACCESS_KEY);
+ propertiesBuilder.put("s3.secret-access-key", SECRET_KEY);
+ }
+
+ restCatalog.initialize("polaris", propertiesBuilder.buildKeepingLast());
+ return restCatalog;
+ }
+
+ protected LoadTableResponse doTestCreateTable(
+ boolean pathStyle, Optional<AccessDelegationMode> dm, boolean stsEnabled) throws IOException {
+ try (var restCatalog =
+ createCatalog(
+ Optional.of(endpoint),
+ stsEnabled ? Optional.of(endpoint) : Optional.empty(),
+ pathStyle,
+ Optional.empty(),
+ dm,
+ stsEnabled,
+ regionForSts(stsEnabled),
+ roleArnForSts(stsEnabled),
+ kmsUnavailableForSts(stsEnabled))) {
+ var loadTableResponse = doTestCreateTable(restCatalog, dm);
+ if (pathStyle) {
+ assertThat(loadTableResponse.config())
+ .containsEntry("s3.path-style-access", Boolean.TRUE.toString());
+ }
+ return loadTableResponse;
+ }
+ }
+
+ protected Optional<String> regionForSts(boolean stsEnabled) {
+ return stsEnabled ? Optional.of(TEST_REGION) : Optional.empty();
+ }
+
+ protected Optional<String> roleArnForSts(boolean stsEnabled) {
+ return stsEnabled ? Optional.of(roleArn) : Optional.empty();
+ }
+
+ protected Optional<Boolean> kmsUnavailableForSts(boolean stsEnabled) {
+ return stsEnabled ? Optional.of(kmsUnavailableForStsCatalogs()) : Optional.empty();
+ }
+
+ protected LoadTableResponse doTestCreateTable(
+ RESTCatalog restCatalog, Optional<AccessDelegationMode> dm) {
+ var tableIdentifier = createTableAndVerifyMetadata(restCatalog);
+ assertThat(restCatalog.tableExists(tableIdentifier)).isTrue();
+
+ try {
+ var loadTableResponse =
+ catalogApi.loadTable(
+ catalogName,
+ tableIdentifier,
+ "ALL",
+ dm.map(v -> Map.of("X-Iceberg-Access-Delegation", v.protocolValue()))
+ .orElse(Map.of()));
+
+ assertThat(loadTableResponse.config()).containsKey(ENDPOINT);
+ return loadTableResponse;
+ } finally {
+ if (restCatalog.tableExists(tableIdentifier)) {
+ restCatalog.dropTable(tableIdentifier);
+ assertThat(restCatalog.tableExists(tableIdentifier)).isFalse();
+ }
+ }
+ }
+
+ protected TableIdentifier createTableAndVerifyMetadata(RESTCatalog restCatalog) {
+ catalogApi.createNamespace(catalogName, "test-ns");
+ var tableIdentifier = TableIdentifier.of("test-ns", "t1");
+ var table = restCatalog.createTable(tableIdentifier, SCHEMA);
+ assertThat(table).isNotNull();
+
+ var ops = ((HasTableOperations) table).operations();
+ var location = URI.create(ops.current().metadataFileLocation());
+
+ var response =
+ s3Client
+ .getObject(
+ GetObjectRequest.builder()
+ .bucket(location.getAuthority())
+ .key(location.getPath().substring(1))
+ .build())
+ .response();
+ assertThat(response.contentLength()).isGreaterThan(0);
+ return tableIdentifier;
+ }
+
+ protected void assertLoadTableWithVendedCredentialsSucceeds(TableIdentifier id) {
+ var response =
+ catalogApi.loadTable(
+ catalogName,
+ id,
+ "ALL",
+ Map.of("X-Iceberg-Access-Delegation", VENDED_CREDENTIALS.protocolValue()));
+ assertThat(response.config())
+ .containsEntry(
+ REFRESH_CREDENTIALS_ENDPOINT,
+ "v1/" + catalogName + "/namespaces/test-ns/tables/t1/credentials");
+ assertThat(response.credentials()).hasSize(1);
+ }
+
+ protected String readS3Object(URI location) throws IOException {
+ try (var is =
+ s3Client.getObject(
+ GetObjectRequest.builder()
+ .bucket(location.getAuthority())
+ .key(location.getPath().substring(1))
+ .build())) {
+ return new String(is.readAllBytes(), UTF_8);
+ }
+ }
+
+ private static void createRoleWithAccess(FlociAwsAccess flociAws) {
+ flociAws
+ .iamClient()
+ .createRole(b -> b.roleName(roleName).assumeRolePolicyDocument(trustPolicy(flociAws)));
+ flociAws
+ .iamClient()
+ .putRolePolicy(
+ b ->
+ b.roleName(roleName)
+ .policyName("polaris-s3-kms-access")
+ .policyDocument(s3AndKmsPolicy(flociAws)));
+ }
+
+ private static String trustPolicy(FlociAwsAccess flociAws) {
+ return """
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Principal": {
+ "AWS": "arn:aws:iam::%s:root"
+ },
+ "Action": "sts:AssumeRole"
+ }
+ ]
+ }
+ """
+ .formatted(flociAws.accountId());
+ }
+
+ private static String s3AndKmsPolicy(FlociAwsAccess flociAws) {
+ return """
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Action": [
+ "s3:GetObject",
+ "s3:PutObject",
+ "s3:DeleteObject"
+ ],
+ "Resource": "arn:aws:s3:::%s/*"
+ },
+ {
+ "Effect": "Allow",
+ "Action": [
+ "s3:ListBucket",
+ "s3:GetBucketLocation"
+ ],
+ "Resource": "arn:aws:s3:::%s"
+ },
+ {
+ "Effect": "Allow",
+ "Action": [
+ "kms:DescribeKey",
+ "kms:Decrypt",
+ "kms:Encrypt",
+ "kms:GenerateDataKey",
+ "kms:GenerateDataKeyWithoutPlaintext"
+ ],
+ "Resource": "arn:aws:kms:%s:%s:key/*"
+ }
+ ]
+ }
+ """
+ .formatted(
+ flociAws.bucket(),
+ flociAws.bucket(),
+ flociAws.region().orElse(TEST_REGION),
+ flociAws.accountId());
+ }
+}
diff --git a/runtime/service/src/intTest/java/org/apache/polaris/service/it/PolarisRestCatalogFlociS3IT.java b/runtime/service/src/intTest/java/org/apache/polaris/service/it/PolarisRestCatalogFlociS3IT.java
new file mode 100644
index 0000000000..c1955cbd50
--- /dev/null
+++ b/runtime/service/src/intTest/java/org/apache/polaris/service/it/PolarisRestCatalogFlociS3IT.java
@@ -0,0 +1,158 @@
+/*
+ * 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 static org.apache.polaris.test.commons.MinioRustProfile.ACCESS_KEY;
+import static org.apache.polaris.test.commons.MinioRustProfile.SECRET_KEY;
+
+import com.google.common.collect.ImmutableMap;
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.common.ResourceArg;
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import io.quarkus.test.junit.TestProfile;
+import java.util.List;
+import java.util.UUID;
+import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.storage.StorageAccessProperty;
+import org.apache.polaris.service.it.env.RestCatalogConfig;
+import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension;
+import org.apache.polaris.service.it.test.PolarisRestCatalogIntegrationBase;
+import org.apache.polaris.test.commons.MinioRustProfile;
+import org.apache.polaris.test.floci.aws.FlociAws;
+import org.apache.polaris.test.floci.aws.FlociAwsAccess;
+import org.apache.polaris.test.floci.aws.FlociAwsTestResource;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@QuarkusIntegrationTest
+@TestProfile(MinioRustProfile.class)
+@QuarkusTestResource(
+ value = FlociAwsTestResource.class,
+ initArgs = {
+ @ResourceArg(name = "accessKey", value = ACCESS_KEY),
+ @ResourceArg(name = "secretKey", value = SECRET_KEY),
+ @ResourceArg(name = "bucket", value = "floci-s3-test-polaris"),
+ @ResourceArg(name = "region", value = PolarisRestCatalogFlociS3IT.TEST_REGION),
+ @ResourceArg(name = "iamEnforcement", value = "true")
+ })
+@ExtendWith(PolarisIntegrationTestExtension.class)
+@RestCatalogConfig({"header.X-Iceberg-Access-Delegation", "vended-credentials"})
+public class PolarisRestCatalogFlociS3IT extends PolarisRestCatalogIntegrationBase {
+
+ protected static final String BUCKET_URI_PREFIX = "/floci-s3-test-polaris";
+ protected static final String TEST_REGION = "us-east-1";
+
+ @FlociAws static FlociAwsAccess flociAwsAccess;
+
+ private static String roleArn;
+ private static String roleSetupKey;
+
+ @Override
+ protected ImmutableMap.Builder<String, String> clientFileIOProperties() {
+ return super.clientFileIOProperties()
+ .put(StorageAccessProperty.AWS_ENDPOINT.getPropertyName(), flociAwsAccess.s3endpoint())
+ .put(StorageAccessProperty.AWS_PATH_STYLE_ACCESS.getPropertyName(), "true")
+ .put(StorageAccessProperty.AWS_KEY_ID.getPropertyName(), ACCESS_KEY)
+ .put(StorageAccessProperty.AWS_SECRET_KEY.getPropertyName(), SECRET_KEY);
+ }
+
+ @Override
+ protected StorageConfigInfo getStorageConfigInfo() {
+ AwsStorageConfigInfo.Builder storageConfig =
+ AwsStorageConfigInfo.builder()
+ .setStorageType(StorageConfigInfo.StorageTypeEnum.S3)
+ .setPathStyleAccess(true)
+ .setEndpoint(flociAwsAccess.s3endpoint())
+ .setStsEndpoint(flociAwsAccess.stsEndpoint())
+ .setRegion(TEST_REGION)
+ .setRoleArn(roleArn())
+ .setKmsUnavailable(true)
+ .setAllowedLocations(List.of(flociAwsAccess.s3BucketUri(BUCKET_URI_PREFIX).toString()));
+
+ return storageConfig.build();
+ }
+
+ private static String roleArn() {
+ String setupKey =
+ flociAwsAccess.endpoint()
+ + "|"
+ + flociAwsAccess.accountId()
+ + "|"
+ + flociAwsAccess.bucket();
+ if (roleArn == null || !setupKey.equals(roleSetupKey)) {
+ String roleName = "polaris-floci-rest-catalog-" + UUID.randomUUID();
+ roleArn = "arn:aws:iam::" + flociAwsAccess.accountId() + ":role/" + roleName;
+ flociAwsAccess
+ .iamClient()
+ .createRole(b -> b.roleName(roleName).assumeRolePolicyDocument(trustPolicy()));
+ flociAwsAccess
+ .iamClient()
+ .putRolePolicy(
+ b -> b.roleName(roleName).policyName("polaris-s3-access").policyDocument(s3Policy()));
+ roleSetupKey = setupKey;
+ }
+ return roleArn;
+ }
+
+ private static String trustPolicy() {
+ return """
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Principal": {
+ "AWS": "arn:aws:iam::%s:root"
+ },
+ "Action": "sts:AssumeRole"
+ }
+ ]
+ }
+ """
+ .formatted(flociAwsAccess.accountId());
+ }
+
+ private static String s3Policy() {
+ return """
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Action": [
+ "s3:GetObject",
+ "s3:PutObject",
+ "s3:DeleteObject"
+ ],
+ "Resource": "arn:aws:s3:::%s/*"
+ },
+ {
+ "Effect": "Allow",
+ "Action": [
+ "s3:ListBucket",
+ "s3:GetBucketLocation"
+ ],
+ "Resource": "arn:aws:s3:::%s"
+ }
+ ]
+ }
+ """
+ .formatted(flociAwsAccess.bucket(), flociAwsAccess.bucket());
+ }
+}
diff --git a/runtime/service/src/intTest/java/org/apache/polaris/service/it/RestCatalogFlociS3SpecialIT.java b/runtime/service/src/intTest/java/org/apache/polaris/service/it/RestCatalogFlociS3SpecialIT.java
new file mode 100644
index 0000000000..563986aee4
--- /dev/null
+++ b/runtime/service/src/intTest/java/org/apache/polaris/service/it/RestCatalogFlociS3SpecialIT.java
@@ -0,0 +1,285 @@
+/*
+ * 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 static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.iceberg.aws.AwsClientProperties.REFRESH_CREDENTIALS_ENDPOINT;
+import static org.apache.iceberg.aws.s3.S3FileIOProperties.ACCESS_KEY_ID;
+import static org.apache.iceberg.aws.s3.S3FileIOProperties.ENDPOINT;
+import static org.apache.iceberg.aws.s3.S3FileIOProperties.SECRET_ACCESS_KEY;
+import static org.apache.polaris.service.catalog.AccessDelegationMode.VENDED_CREDENTIALS;
+import static org.apache.polaris.test.commons.MinioRustProfile.ACCESS_KEY;
+import static org.apache.polaris.test.commons.MinioRustProfile.SECRET_KEY;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.common.ResourceArg;
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import io.quarkus.test.junit.TestProfile;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
+import org.apache.polaris.service.catalog.AccessDelegationMode;
+import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension;
+import org.apache.polaris.test.commons.MinioRustProfile;
+import org.apache.polaris.test.floci.aws.FlociAws;
+import org.apache.polaris.test.floci.aws.FlociAwsAccess;
+import org.apache.polaris.test.floci.aws.FlociAwsTestResource;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * These tests complement {@link PolarisRestCatalogFlociS3IT} to validate client-side access to
+ * Floci S3 storage via {@code FileIO} instances configured from catalog's {@code loadTable}
+ * responses with some S3-specific options.
+ */
+@QuarkusIntegrationTest
+@TestProfile(MinioRustProfile.class)
+@QuarkusTestResource(
+ value = FlociAwsTestResource.class,
+ initArgs = {
+ @ResourceArg(name = "accessKey", value = ACCESS_KEY),
+ @ResourceArg(name = "secretKey", value = SECRET_KEY),
+ @ResourceArg(name = "bucket", value = "floci-s3-special-test"),
+ @ResourceArg(name = "region", value = AbstractRestCatalogFlociS3SpecialIT.TEST_REGION),
+ @ResourceArg(name = "iamEnforcement", value = "true")
+ })
+@ExtendWith(PolarisIntegrationTestExtension.class)
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class RestCatalogFlociS3SpecialIT extends AbstractRestCatalogFlociS3SpecialIT {
+
+ @FlociAws static FlociAwsAccess flociAwsAccess;
+
+ @Override
+ protected FlociAwsAccess flociAwsAccess() {
+ return flociAwsAccess;
+ }
+
+ @ParameterizedTest
+ @CsvSource({"true, true,", "false, true,", "true, false,", "false, false,"})
+ public void testCreateTable(boolean pathStyle, boolean stsEnabled) throws IOException {
+ var response = doTestCreateTable(pathStyle, Optional.empty(), stsEnabled);
+ assertThat(response.config()).doesNotContainKey(SECRET_ACCESS_KEY);
+ assertThat(response.config()).doesNotContainKey(ACCESS_KEY_ID);
+ assertThat(response.config()).doesNotContainKey(REFRESH_CREDENTIALS_ENDPOINT);
+ assertThat(response.credentials()).isEmpty();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testCreateTableVendedCredentials(boolean pathStyle) throws IOException {
+ var response = doTestCreateTable(pathStyle, Optional.of(VENDED_CREDENTIALS), true);
+ assertThat(response.config())
+ .containsEntry(
+ REFRESH_CREDENTIALS_ENDPOINT,
+ "v1/" + catalogName + "/namespaces/test-ns/tables/t1/credentials");
+ assertThat(response.credentials()).hasSize(1);
+ }
+
+ @Test
+ public void testCreateTableVendedCredentialsWithoutRegionPasses() throws IOException {
+ try (var restCatalog =
+ createCatalog(
+ Optional.of(endpoint),
+ Optional.of(endpoint),
+ true,
+ Optional.empty(),
+ Optional.of(VENDED_CREDENTIALS),
+ true,
+ Optional.empty(),
+ Optional.of(roleArn),
+ Optional.of(kmsUnavailableForStsCatalogs()))) {
+ var id = createTableAndVerifyMetadata(restCatalog);
+ try {
+ assertLoadTableWithVendedCredentialsSucceeds(id);
+ } finally {
+ catalogApi.dropTable(catalogName, id);
+ }
+ }
+ }
+
+ @Test
+ public void testCreateTableVendedCredentialsWithFullAwsShapePasses() throws IOException {
+ try (var restCatalog =
+ createCatalog(
+ Optional.of(endpoint),
+ Optional.of(endpoint),
+ true,
+ Optional.empty(),
+ Optional.of(VENDED_CREDENTIALS),
+ true,
+ Optional.of(TEST_REGION),
+ Optional.of(roleArn),
+ Optional.of(kmsUnavailableForStsCatalogs()))) {
+ var id = createTableAndVerifyMetadata(restCatalog);
+ try {
+ assertLoadTableWithVendedCredentialsSucceeds(id);
+ } finally {
+ catalogApi.dropTable(catalogName, id);
+ }
+ }
+ }
+
+ @Test
+ public void testInternalEndpoints() throws IOException {
+ try (var restCatalog =
+ createCatalog(
+ Optional.of("http://s3.example.com"),
+ Optional.of(endpoint),
+ false,
+ Optional.of(endpoint),
+ Optional.empty(),
+ true,
+ regionForSts(true),
+ roleArnForSts(true),
+ kmsUnavailableForSts(true))) {
+ var storageConfig = managementApi.getCatalog(catalogName).getStorageConfigInfo();
+ assertThat((AwsStorageConfigInfo) storageConfig)
+ .extracting(
+ AwsStorageConfigInfo::getEndpoint,
+ AwsStorageConfigInfo::getStsEndpoint,
+ AwsStorageConfigInfo::getEndpointInternal,
+ AwsStorageConfigInfo::getPathStyleAccess)
+ .containsExactly("http://s3.example.com", endpoint, endpoint, false);
+ var loadTableResponse = doTestCreateTable(restCatalog, Optional.empty());
+ assertThat(loadTableResponse.config()).containsEntry(ENDPOINT, "http://s3.example.com");
+ }
+ }
+
+ @Test
+ public void testCreateTableFailureWithCredentialVendingWithoutSts() throws IOException {
+ try (var restCatalog =
+ createCatalog(
+ Optional.of(endpoint),
+ Optional.of("http://sts.example.com"),
+ false,
+ Optional.of(VENDED_CREDENTIALS),
+ false)) {
+ var storageConfig = managementApi.getCatalog(catalogName).getStorageConfigInfo();
+ assertThat((AwsStorageConfigInfo) storageConfig)
+ .extracting(
+ AwsStorageConfigInfo::getEndpoint,
+ AwsStorageConfigInfo::getStsEndpoint,
+ AwsStorageConfigInfo::getEndpointInternal,
+ AwsStorageConfigInfo::getPathStyleAccess,
+ AwsStorageConfigInfo::getStsUnavailable)
+ .containsExactly(endpoint, "http://sts.example.com", null, false, true);
+
+ catalogApi.createNamespace(catalogName, "test-ns");
+ var id = TableIdentifier.of("test-ns", "t2");
+ assertThatThrownBy(() -> restCatalog.createTable(id, SCHEMA))
+ .hasMessageContaining("but no credentials are available")
+ .hasMessageContaining(id.toString());
+ }
+ }
+
+ @Test
+ public void testLoadTableFailureWithCredentialVendingWithoutSts() throws IOException {
+ try (var restCatalog =
+ createCatalog(
+ Optional.of(endpoint),
+ Optional.of("http://sts.example.com"),
+ false,
+ Optional.empty(),
+ false)) {
+
+ catalogApi.createNamespace(catalogName, "test-ns");
+ var id = TableIdentifier.of("test-ns", "t3");
+ restCatalog.createTable(id, SCHEMA);
+
+ assertThatThrownBy(
+ () ->
+ catalogApi.loadTable(
+ catalogName,
+ id,
+ "ALL",
+ Map.of("X-Iceberg-Access-Delegation", VENDED_CREDENTIALS.protocolValue())))
+ .hasMessageContaining("but no credentials are available")
+ .hasMessageContaining(id.toString());
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "true, true,",
+ "false, true,",
+ "true, false,",
+ "false, false,",
+ "true, true, VENDED_CREDENTIALS",
+ "false, true, VENDED_CREDENTIALS"
+ })
+ public void testAppendFiles(
+ boolean pathStyle, boolean stsEnabled, AccessDelegationMode delegationMode)
+ throws IOException {
+ try (var restCatalog =
+ createCatalog(
+ Optional.of(endpoint),
+ Optional.of(endpoint),
+ pathStyle,
+ Optional.empty(),
+ Optional.ofNullable(delegationMode),
+ stsEnabled,
+ regionForSts(stsEnabled),
+ roleArnForSts(stsEnabled),
+ kmsUnavailableForSts(stsEnabled))) {
+ catalogApi.createNamespace(catalogName, "test-ns");
+ var id = TableIdentifier.of("test-ns", "t1");
+ var table = restCatalog.createTable(id, SCHEMA);
+ assertThat(table).isNotNull();
+
+ @SuppressWarnings("resource")
+ var io = table.io();
+
+ var loc =
+ URI.create(
+ table
+ .locationProvider()
+ .newDataLocation(
+ String.format(
+ "test-file-%s-%s-%s.txt", pathStyle, delegationMode, stsEnabled)));
+ var f1 = io.newOutputFile(loc.toString());
+ try (var os = f1.create()) {
+ os.write("Hello World".getBytes(UTF_8));
+ }
+
+ var df =
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath(f1.location())
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(4)
+ .withRecordCount(1)
+ .build();
+
+ table.newAppend().appendFile(df).commit();
+
+ assertThat(readS3Object(loc)).isEqualTo("Hello World");
+ }
+ }
+}
diff --git a/runtime/service/src/intTest/java/org/apache/polaris/service/it/RestCatalogFlociS3SpecialKmsIT.java b/runtime/service/src/intTest/java/org/apache/polaris/service/it/RestCatalogFlociS3SpecialKmsIT.java
new file mode 100644
index 0000000000..9370eb500a
--- /dev/null
+++ b/runtime/service/src/intTest/java/org/apache/polaris/service/it/RestCatalogFlociS3SpecialKmsIT.java
@@ -0,0 +1,57 @@
+/*
+ * 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 static org.apache.polaris.test.commons.MinioRustProfile.ACCESS_KEY;
+import static org.apache.polaris.test.commons.MinioRustProfile.SECRET_KEY;
+
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.common.ResourceArg;
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import io.quarkus.test.junit.TestProfile;
+import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension;
+import org.apache.polaris.test.commons.MinioRustProfile;
+import org.apache.polaris.test.floci.aws.FlociAwsTestResource;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Validates that Floci can exercise Polaris credential vending when AWS KMS policy generation is
+ * enabled.
+ */
+@QuarkusIntegrationTest
+@TestProfile(MinioRustProfile.class)
+@QuarkusTestResource(
+ value = FlociAwsTestResource.class,
+ initArgs = {
+ @ResourceArg(name = "accessKey", value = ACCESS_KEY),
+ @ResourceArg(name = "secretKey", value = SECRET_KEY),
+ @ResourceArg(name = "bucket", value = "floci-s3-special-kms-test"),
+ @ResourceArg(name = "region", value = AbstractRestCatalogFlociS3SpecialIT.TEST_REGION),
+ @ResourceArg(name = "iamEnforcement", value = "true")
+ })
+@ExtendWith(PolarisIntegrationTestExtension.class)
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class RestCatalogFlociS3SpecialKmsIT extends RestCatalogFlociS3SpecialIT {
+
+ @Override
+ protected boolean kmsUnavailableForStsCatalogs() {
+ return false;
+ }
+}