You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@polaris.apache.org by di...@apache.org on 2026/10/05 21:27:22 UTC
(polaris) branch main updated: Add S3CredentialVendingMechanism for S3 catalogs (#5513)
This is an automated email from the ASF dual-hosted git repository.
dimas-b 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 d9b44e1ff6 Add S3CredentialVendingMechanism for S3 catalogs (#5513)
d9b44e1ff6 is described below
commit d9b44e1ff6dea5cc03b329ad173d7e8a3dfb461b
Author: austent <au...@gmail.com>
AuthorDate: Mon Oct 5 16:27:17 2026 -0500
Add S3CredentialVendingMechanism for S3 catalogs (#5513)
* Add credentialVendingMechanism to AwsStorageConfigurationInfo
* Expose credentialVendingMechanism on AwsStorageConfigInfo and map it in CatalogEntity
* Check credentialVendingMechanism at catalog initialization, storage access and the provider
---------
Co-authored-by: Austen Tomek <au...@chicagotrading.com>
---
CHANGELOG.md | 19 +
.../polaris/core/config/FeatureConfiguration.java | 18 +
.../apache/polaris/core/entity/CatalogEntity.java | 4 +
.../storage/aws/AwsStorageConfigurationInfo.java | 33 ++
.../storage/aws/S3CredentialVendingMechanism.java | 56 ++
.../PolarisStorageConfigurationInfoTest.java | 10 +
.../aws/AwsStorageConfigurationInfoTest.java | 78 +++
.../polaris/service/admin/PolarisAdminService.java | 50 +-
.../validation/IcebergPropertiesValidation.java | 18 +
.../service/config/ProductionReadinessChecks.java | 76 +++
.../PolarisStorageIntegrationProviderImpl.java | 51 +-
.../storage/S3CredentialVendingMechanisms.java | 127 +++++
.../storage/StsCredentialVendingMechanism.java | 97 ++++
.../service/admin/ManagementServiceTest.java | 484 ++++++++++++++++-
.../admin/PolarisAdminServiceAuthzTest.java | 14 +-
.../service/admin/PolarisAdminServiceTest.java | 87 +++-
.../admin/PolarisAdminServiceTestSupport.java | 21 +-
.../service/admin/PolarisAuthzTestBase.java | 11 +-
.../service/admin/PolarisServiceImplTest.java | 58 ++-
.../S3CredentialVendingMechanismRoutesTest.java | 360 +++++++++++++
.../service/admin/StorageConfigInfoJsonTest.java | 109 ++++
.../policy/PolicyCatalogHandlerAuthzTest.java | 7 +
.../IcebergPropertiesValidationTest.java | 93 ++++
.../config/ProductionReadinessChecksTest.java | 118 +++++
.../polaris/service/entity/CatalogEntityTest.java | 27 +
.../RecordingStsCredentialVendingMechanism.java | 70 +++
.../storage/RecordingStsMechanismProfile.java | 55 ++
.../RecordingThirdCredentialVendingMechanism.java | 96 ++++
.../S3CredentialVendingMechanismCdiTest.java | 572 +++++++++++++++++++++
.../S3CredentialVendingMechanismDispatchTest.java | 174 +++++++
...edentialVendingMechanismStsOverrideCdiTest.java | 150 ++++++
...ntialVendingMechanismThirdMechanismCdiTest.java | 243 +++++++++
.../storage/S3CredentialVendingMechanismsTest.java | 198 +++++++
.../storage/StsCredentialVendingMechanismTest.java | 230 +++++++++
.../service/storage/TestInMemoryFileIOFactory.java | 53 ++
.../service/storage/ThirdMechanismProfile.java | 69 +++
.../org/apache/polaris/service/TestServices.java | 53 +-
.../config-sections/flags-polaris_features.md | 9 +
spec/polaris-management-service.yml | 12 +
39 files changed, 3968 insertions(+), 42 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index eeabab9880..9bf8059ade 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -37,8 +37,27 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
are not compatible with internal authentication and internal authorization; you must configure an
external IDP and an external PDP, such as OPA or Ranger.
+- S3 storage configurations gain an optional string field, `credentialVendingMechanism`. A catalog
+ that leaves it empty uses the server's default mechanism, STS AssumeRole, and management API
+ responses omit the field for it; `STS` selects the same mechanism explicitly and is the only value
+ this release supports. Mechanisms are CDI beans discovered by their `@Identifier` at startup,
+ and each one can validate the catalogs that select it at create and update. A realm lists the
+ explicit mechanisms it accepts in the new `SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS` feature
+ (default `[STS]`), enforced at catalog create and update and when the server builds the storage
+ integration that vends credentials for a catalog. A catalog that names a mechanism this server
+ does not provide is refused at create and update, and whenever a credential is vended for it,
+ with "S3 credential vending mechanism `<id>` is not available in this server".
+
### Upgrade notes
+- `SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS` lists the explicit mechanisms a realm accepts; an empty
+ `credentialVendingMechanism` is always allowed. Startup reports a listed mechanism with no
+ installed bean as a non-severe readiness warning. Startup fails (a severe readiness issue) when no
+ `STS` mechanism is installed, because every catalog with an empty field depends on it.
+- Rolling upgrades: a node still running an earlier release does not know `credentialVendingMechanism`.
+ It ignores the field in a stored configuration and vends through STS, and it drops the field from
+ the catalog create and update requests it serves. Keep S3 storage configurations unchanged during
+ the upgrade.
- Polaris-managed AWS SDK clients now use Apache HttpClient 5, which disables HTTP
`Expect: 100-continue` by default. Set `polaris.storage.expect-continue-enabled=true`
to preserve the previous behavior. Iceberg S3 clients continue to use Apache HttpClient 4
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 7c527a068f..c5f8dcc145 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
@@ -24,6 +24,7 @@ import org.apache.polaris.core.admin.model.AuthenticationParameters;
import org.apache.polaris.core.admin.model.StorageConfigInfo;
import org.apache.polaris.core.connection.ConnectionType;
import org.apache.polaris.core.persistence.cache.EntityWeigher;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
/**
* Configurations for features within Polaris. These configurations are intended to be customized
@@ -403,6 +404,23 @@ public class FeatureConfiguration<T> extends PolarisConfiguration<T> {
StorageConfigInfo.StorageTypeEnum.GCS.name()))
.buildFeatureConfiguration();
+ public static final FeatureConfiguration<List<String>>
+ SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS =
+ PolarisConfiguration.<List<String>>builder()
+ .key("SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS")
+ .description(
+ "The credential vending mechanisms an S3 catalog in this realm may name explicitly. Names\n"
+ + "are case-sensitive and must match a mechanism's identifier exactly, for example STS.\n"
+ + "Apache Polaris supports STS, which is STS AssumeRole. A catalog that\n"
+ + "leaves credentialVendingMechanism empty uses the server's default mechanism and is\n"
+ + "always allowed. Realm-level only; catalog properties cannot widen it. Enforced at\n"
+ + "catalog create and update, and when the server builds the storage integration that\n"
+ + "vends credentials for a catalog. A listed mechanism with no implementation in this\n"
+ + "server is reported at startup and refused at catalog create and update and whenever\n"
+ + "a credential is vended for a catalog that names it.")
+ .defaultValue(List.of(S3CredentialVendingMechanism.STS))
+ .buildFeatureConfiguration();
+
public static final FeatureConfiguration<Boolean> CLEANUP_ON_NAMESPACE_DROP =
PolarisConfiguration.<Boolean>builder()
.key("CLEANUP_ON_NAMESPACE_DROP")
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java b/polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java
index ce4074816e..d03b9b3ae3 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java
@@ -199,6 +199,7 @@ public class CatalogEntity extends PolarisEntity implements LocationBasedEntity
.setStsUnavailable(awsConfig.getStsUnavailable())
.setEndpointInternal(awsConfig.getEndpointInternal())
.setKmsUnavailable(awsConfig.getKmsUnavailable())
+ .setCredentialVendingMechanism(awsConfig.getCredentialVendingMechanism())
.build();
}
@@ -419,6 +420,9 @@ public class CatalogEntity extends PolarisEntity implements LocationBasedEntity
return AwsStorageConfigurationInfo.builder()
.allowedLocations(allowedLocations)
.storageName(awsConfigModel.getStorageName())
+ .credentialVendingMechanism(
+ AwsStorageConfigurationInfo.credentialVendingMechanismOf(
+ awsConfigModel.getCredentialVendingMechanism()))
.roleARN(awsConfigModel.getRoleArn())
.encryptionKeys(encryptionKeys)
.decryptionKeys(awsConfigModel.getDecryptionKeys())
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfo.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfo.java
index 4a6a9db517..319ef44fa6 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfo.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfo.java
@@ -142,6 +142,35 @@ public abstract class AwsStorageConfigurationInfo extends PolarisStorageConfigur
/** Flag indicating whether path-style bucket access should be forced in S3 clients. */
public abstract @Nullable Boolean getPathStyleAccess();
+ /**
+ * The mechanism that vends credentials for this catalog, an identifier such as {@link
+ * S3CredentialVendingMechanism#STS}, or null when the catalog leaves the choice to the server;
+ * see {@link #resolvedCredentialVendingMechanism()}. A row persisted before this field existed
+ * lacks it and reads as null. The value is carried as is; the realm allowlist, the mechanism
+ * registry and the mechanism itself decide whether it is accepted.
+ */
+ @Nullable
+ public abstract String getCredentialVendingMechanism();
+
+ /**
+ * The identifier to resolve in the mechanism registry: the explicit value, or {@link
+ * S3CredentialVendingMechanism#STS} when the field is empty. The stored value stays empty.
+ */
+ @JsonIgnore
+ public String resolvedCredentialVendingMechanism() {
+ String mechanism = getCredentialVendingMechanism();
+ return mechanism == null ? S3CredentialVendingMechanism.STS : mechanism;
+ }
+
+ /**
+ * Normalizes a value received through the management API: null and blank both mean the field is
+ * empty, so the catalog uses the server's default mechanism, STS.
+ */
+ @Nullable
+ public static String credentialVendingMechanismOf(@Nullable String value) {
+ return value == null || value.isBlank() ? null : value;
+ }
+
/**
* Flag indicating whether STS is available or not. It is modeled in the negative to simplify
* support for unset values ({@code null} being interpreted as {@code false}).
@@ -193,6 +222,10 @@ public abstract class AwsStorageConfigurationInfo extends PolarisStorageConfigur
@Override
protected void check() {
super.check();
+ String mechanism = getCredentialVendingMechanism();
+ if (mechanism != null && mechanism.isBlank()) {
+ throw new IllegalArgumentException("credentialVendingMechanism must not be blank");
+ }
String arn = getRoleARN();
validateArn(arn);
if (arn != null) {
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/S3CredentialVendingMechanism.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/S3CredentialVendingMechanism.java
new file mode 100644
index 0000000000..baadda0c20
--- /dev/null
+++ b/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/S3CredentialVendingMechanism.java
@@ -0,0 +1,56 @@
+/*
+ * 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.core.storage.aws;
+
+import org.apache.polaris.core.storage.PolarisStorageIntegration;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * How Polaris vends S3 credentials for one S3 catalog. Implementations are CDI beans annotated with
+ * {@code @Identifier("<mechanism>")}; a catalog's {@code credentialVendingMechanism} selects one by
+ * that identifier, and a catalog that leaves the field empty selects {@link #STS}. A realm lists
+ * the explicit mechanisms it accepts in {@code SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS}; a
+ * listed mechanism with no bean in the running server is refused at catalog create and update, and
+ * whenever a credential is vended for a catalog that selects it. A server replaces a mechanism,
+ * {@link #STS} included, with an {@code @Alternative} bean of a higher {@code @Priority} that
+ * carries the same identifier. An implementation must be {@code @ApplicationScoped} and
+ * thread-safe: the registry resolves every bean once at startup, before any request context exists,
+ * and one instance serves all concurrent requests for the lifetime of the server. An implementation
+ * that needs realm configuration injects {@code RealmConfig}; the request context is active
+ * wherever the server calls a mechanism, including task execution.
+ */
+public interface S3CredentialVendingMechanism {
+
+ /** STS AssumeRole against the catalog's role. A catalog that leaves the field empty uses it. */
+ String STS = "STS";
+
+ /** The storage integration that vends for one S3 catalog under this mechanism. */
+ PolarisStorageIntegration integrationFor(AwsStorageConfigurationInfo storageConfig);
+
+ /**
+ * Checks a storage config that selects this mechanism before it is stored. Called at catalog
+ * create with {@code current} null, and at catalog update with the stored config as {@code
+ * current}, after authorization, the realm allowlist and the availability check. Throw {@link
+ * IllegalArgumentException} to refuse the request with HTTP 400. The default accepts every
+ * config.
+ */
+ default void validate(
+ @Nullable AwsStorageConfigurationInfo current, AwsStorageConfigurationInfo updated) {}
+}
diff --git a/polaris-core/src/test/java/org/apache/polaris/core/storage/PolarisStorageConfigurationInfoTest.java b/polaris-core/src/test/java/org/apache/polaris/core/storage/PolarisStorageConfigurationInfoTest.java
index d1925873f3..fdf2769767 100644
--- a/polaris-core/src/test/java/org/apache/polaris/core/storage/PolarisStorageConfigurationInfoTest.java
+++ b/polaris-core/src/test/java/org/apache/polaris/core/storage/PolarisStorageConfigurationInfoTest.java
@@ -95,6 +95,16 @@ public class PolarisStorageConfigurationInfoTest {
.storageName("my-storage")
.build(),
"{\"@type\":\"AwsStorageConfigurationInfo\",\"storageType\":\"S3\",\"allowedLocations\":[\"s3://foo/bar\",\"s3://no/where\"],\"storageName\":\"my-storage\",\"roleARN\":\"arn:aws:iam::123456789012:role/polaris-test\",\"region\":\"no-where-1\",\"fileIoImplClassName\":\"org.apache.iceberg.aws.s3.S3FileIO\"}"),
+ arguments(
+ AwsStorageConfigurationInfo.builder()
+ .addAllowedLocations("s3://other-bucket/base/")
+ .credentialVendingMechanism("DOWNSTREAM_MECHANISM")
+ .endpoint("https://s3.example-compatible-store.test")
+ .pathStyleAccess(true)
+ .region("auto")
+ .storageName("other")
+ .build(),
+ "{\"@type\":\"AwsStorageConfigurationInfo\",\"storageType\":\"S3\",\"credentialVendingMechanism\":\"DOWNSTREAM_MECHANISM\",\"allowedLocations\":[\"s3://other-bucket/base/\"],\"storageName\":\"other\",\"region\":\"auto\",\"endpoint\":\"https://s3.example-compatible-store.test\",\"pathStyleAccess\":true,\"fileIoImplClassName\":\"org.apache.iceberg.aws.s3.S3FileIO\"}"),
//
arguments(
GcpStorageConfigurationInfo.builder()
diff --git a/polaris-core/src/test/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfoTest.java b/polaris-core/src/test/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfoTest.java
index 324fffa6cd..a5d758a2bf 100644
--- a/polaris-core/src/test/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfoTest.java
+++ b/polaris-core/src/test/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfoTest.java
@@ -20,10 +20,12 @@
package org.apache.polaris.core.storage.aws;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.net.URI;
import java.util.List;
import java.util.stream.Stream;
+import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -197,4 +199,80 @@ public class AwsStorageConfigurationInfoTest {
"urn:ecs:sts::namespace:assumed-role/s3assumeRole/user1-105-temp", "namespace", "ecs"),
Arguments.of("urn:sgws:identity::12345:group/foo-bar-abcdef", "12345", "sgws"));
}
+
+ @Test
+ void anEmptyMechanismReadsAsNullAndResolvesToSts() {
+ AwsStorageConfigurationInfo config =
+ newBuilder().addAllowedLocation("s3://bucket/path/").build();
+ assertThat(config.getCredentialVendingMechanism()).isNull();
+ assertThat(config.resolvedCredentialVendingMechanism())
+ .isEqualTo(S3CredentialVendingMechanism.STS);
+ assertThat(config.serialize()).doesNotContain("credentialVendingMechanism");
+ }
+
+ @Test
+ void anExplicitMechanismResolvesToItself() {
+ AwsStorageConfigurationInfo config =
+ newBuilder()
+ .addAllowedLocation("s3://bucket/path/")
+ .credentialVendingMechanism(S3CredentialVendingMechanism.STS)
+ .build();
+ assertThat(config.resolvedCredentialVendingMechanism())
+ .isEqualTo(S3CredentialVendingMechanism.STS);
+ assertThat(config.serialize()).contains("\"credentialVendingMechanism\":\"STS\"");
+ }
+
+ @Test
+ void aPersistedRowWithoutTheKeyReadsAsEmpty() throws Exception {
+ String json =
+ "{\"@type\":\"AwsStorageConfigurationInfo\",\"storageType\":\"S3\","
+ + "\"allowedLocations\":[\"s3://bucket/\"]}";
+ AwsStorageConfigurationInfo info =
+ (AwsStorageConfigurationInfo) PolarisStorageConfigurationInfo.deserialize(json);
+ assertThat(info.getCredentialVendingMechanism()).isNull();
+ assertThat(info.resolvedCredentialVendingMechanism())
+ .isEqualTo(S3CredentialVendingMechanism.STS);
+ }
+
+ @Test
+ void theApiValueNormalizerTreatsNullAndBlankAsEmpty() {
+ assertThat(AwsStorageConfigurationInfo.credentialVendingMechanismOf(null)).isNull();
+ assertThat(AwsStorageConfigurationInfo.credentialVendingMechanismOf("")).isNull();
+ assertThat(AwsStorageConfigurationInfo.credentialVendingMechanismOf(" ")).isNull();
+ assertThat(AwsStorageConfigurationInfo.credentialVendingMechanismOf("STS")).isEqualTo("STS");
+ }
+
+ @Test
+ void blankMechanismIsRejected() {
+ assertThatThrownBy(
+ () ->
+ AwsStorageConfigurationInfo.builder()
+ .credentialVendingMechanism(" ")
+ .addAllowedLocation("s3://bucket/")
+ .build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("credentialVendingMechanism must not be blank");
+ }
+
+ @Test
+ void unknownMechanismIsCarriedByTheModel() {
+ assertThat(
+ AwsStorageConfigurationInfo.builder()
+ .credentialVendingMechanism("DOWNSTREAM_MECHANISM")
+ .addAllowedLocation("s3://bucket/")
+ .build()
+ .getCredentialVendingMechanism())
+ .isEqualTo("DOWNSTREAM_MECHANISM");
+ }
+
+ @Test
+ void aPersistedRowWithTheKeyReadsItBack() throws Exception {
+ String json =
+ "{\"@type\":\"AwsStorageConfigurationInfo\",\"storageType\":\"S3\","
+ + "\"allowedLocations\":[\"s3://bucket/\"],"
+ + "\"credentialVendingMechanism\":\"DOWNSTREAM_MECHANISM\"}";
+ AwsStorageConfigurationInfo info =
+ (AwsStorageConfigurationInfo) PolarisStorageConfigurationInfo.deserialize(json);
+ assertThat(info.getCredentialVendingMechanism()).isEqualTo("DOWNSTREAM_MECHANISM");
+ }
}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java
index b89326774f..227bad2619 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java
@@ -48,6 +48,7 @@ import org.apache.iceberg.exceptions.ValidationException;
import org.apache.polaris.core.PolarisCallContext;
import org.apache.polaris.core.StructuredLogKeys;
import org.apache.polaris.core.admin.model.AuthenticationParameters;
+import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
import org.apache.polaris.core.admin.model.BearerAuthenticationParameters;
import org.apache.polaris.core.admin.model.Catalog;
import org.apache.polaris.core.admin.model.CatalogGrant;
@@ -69,6 +70,7 @@ import org.apache.polaris.core.admin.model.PrincipalWithCredentialsCredentials;
import org.apache.polaris.core.admin.model.ResetPrincipalRequest;
import org.apache.polaris.core.admin.model.SemanticModelGrant;
import org.apache.polaris.core.admin.model.SemanticModelPrivilege;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
import org.apache.polaris.core.admin.model.TableGrant;
import org.apache.polaris.core.admin.model.TablePrivilege;
import org.apache.polaris.core.admin.model.UpdateCatalogRequest;
@@ -138,9 +140,12 @@ import org.apache.polaris.core.semantic.exceptions.NoSuchSemanticModelException;
import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
import org.apache.polaris.core.storage.StorageLocation;
import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
import org.apache.polaris.core.storage.azure.AzureStorageConfigurationInfo;
import org.apache.polaris.service.catalog.common.PolarisSecurableMapper;
+import org.apache.polaris.service.catalog.validation.IcebergPropertiesValidation;
import org.apache.polaris.service.config.ReservedProperties;
+import org.apache.polaris.service.storage.S3CredentialVendingMechanisms;
import org.apache.polaris.service.types.PolicyIdentifier;
import org.jspecify.annotations.NonNull;
import org.jspecify.annotations.Nullable;
@@ -170,6 +175,7 @@ public class PolarisAdminService {
private final UserSecretsManager userSecretsManager;
private final ServiceIdentityProvider serviceIdentityProvider;
private final ReservedProperties reservedProperties;
+ private final S3CredentialVendingMechanisms vendingMechanisms;
@Inject
public PolarisAdminService(
@@ -180,7 +186,8 @@ public class PolarisAdminService {
@NonNull ServiceIdentityProvider serviceIdentityProvider,
@NonNull PolarisPrincipal principal,
@NonNull PolarisAuthorizer authorizer,
- @NonNull ReservedProperties reservedProperties) {
+ @NonNull ReservedProperties reservedProperties,
+ @NonNull S3CredentialVendingMechanisms vendingMechanisms) {
this.callContext = callContext;
this.realmConfig = callContext.getRealmConfig();
this.resolutionManifestFactory = resolutionManifestFactory;
@@ -190,6 +197,7 @@ public class PolarisAdminService {
this.userSecretsManager = userSecretsManager;
this.serviceIdentityProvider = serviceIdentityProvider;
this.reservedProperties = reservedProperties;
+ this.vendingMechanisms = vendingMechanisms;
}
private PolarisCallContext getCurrentPolarisContext() {
@@ -825,8 +833,15 @@ public class PolarisAdminService {
public PolarisEntity createCatalog(CreateCatalogRequest catalogRequest) {
authorizeBasicRootOperationOrThrow(PolarisAuthorizableOperation.CREATE_CATALOG);
Catalog catalog = catalogRequest.getCatalog();
+ S3CredentialVendingMechanism mechanism =
+ validateS3CredentialVendingMechanism(catalog.getStorageConfigInfo());
CatalogEntity entity = CatalogEntity.fromCatalog(realmConfig, catalog);
+ if (mechanism != null) {
+ // Safe: CatalogEntity.Builder.processStorageConfigurationInfo always converts an
+ // AwsStorageConfigInfo request to an AwsStorageConfigurationInfo.
+ mechanism.validate(null, (AwsStorageConfigurationInfo) entity.getStorageConfigurationInfo());
+ }
checkArgument(entity.getId() == -1, "Entity to be created must have no ID assigned");
@@ -964,6 +979,24 @@ public class PolarisAdminService {
return getCatalogByName(resolutionManifest, name);
}
+ /**
+ * The mechanism checks that run after authorization, so an unauthorized caller learns nothing
+ * about the realm's configuration: the realm allowlist for an explicit value, then availability
+ * in this server. Returns the mechanism the catalog selects, or null for a storage config that is
+ * not S3.
+ */
+ private @Nullable S3CredentialVendingMechanism validateS3CredentialVendingMechanism(
+ @Nullable StorageConfigInfo storageConfigInfo) {
+ if (!(storageConfigInfo instanceof AwsStorageConfigInfo s3Config)) {
+ return null;
+ }
+ String requested = s3Config.getCredentialVendingMechanism();
+ String explicit = AwsStorageConfigurationInfo.credentialVendingMechanismOf(requested);
+ IcebergPropertiesValidation.validateS3CredentialVendingMechanismAllowed(realmConfig, explicit);
+ return vendingMechanisms.require(
+ explicit == null ? S3CredentialVendingMechanism.STS : explicit);
+ }
+
/**
* Helper to validate business logic of what is allowed to be updated or throw a
* BadRequestException.
@@ -1021,6 +1054,8 @@ public class PolarisAdminService {
PolarisAuthorizableOperation op = PolarisAuthorizableOperation.UPDATE_CATALOG;
PolarisResolutionManifest resolutionManifest =
authorizeBasicTopLevelEntityOperationOrThrow(op, name, PolarisEntityType.CATALOG);
+ S3CredentialVendingMechanism mechanism =
+ validateS3CredentialVendingMechanism(updateRequest.getStorageConfigInfo());
CatalogEntity currentCatalogEntity = getCatalogByName(resolutionManifest, name);
@@ -1057,6 +1092,19 @@ public class PolarisAdminService {
validateUpdateCatalogDiffOrThrow(currentCatalogEntity, updatedEntity);
+ if (mechanism != null) {
+ PolarisStorageConfigurationInfo currentStorageConfig =
+ currentCatalogEntity.getStorageConfigurationInfo();
+ // Safe: CatalogEntity.Builder.processStorageConfigurationInfo always converts an
+ // AwsStorageConfigInfo request to an AwsStorageConfigurationInfo, and the storage-type
+ // freeze above has already refused a type change.
+ mechanism.validate(
+ currentStorageConfig instanceof AwsStorageConfigurationInfo currentAwsConfig
+ ? currentAwsConfig
+ : null,
+ (AwsStorageConfigurationInfo) updatedEntity.getStorageConfigurationInfo());
+ }
+
if (catalogOverlapsWithExistingCatalog(updatedEntity)) {
throw new ValidationException(
"Cannot update Catalog %s. One or more of its new locations overlaps with an existing catalog",
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/validation/IcebergPropertiesValidation.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/validation/IcebergPropertiesValidation.java
index 790433035d..41e0179b89 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/validation/IcebergPropertiesValidation.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/validation/IcebergPropertiesValidation.java
@@ -21,7 +21,9 @@ package org.apache.polaris.service.catalog.validation;
import static org.apache.polaris.core.config.FeatureConfiguration.ALLOW_INSECURE_STORAGE_TYPES;
import static org.apache.polaris.core.config.FeatureConfiguration.ALLOW_SPECIFYING_FILE_IO_IMPL;
import static org.apache.polaris.core.config.FeatureConfiguration.SUPPORTED_CATALOG_STORAGE_TYPES;
+import static org.apache.polaris.core.config.FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS;
+import java.util.List;
import java.util.Map;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.exceptions.ValidationException;
@@ -83,6 +85,22 @@ public class IcebergPropertiesValidation {
return ioImpl;
}
+ /**
+ * The realm allowlist for an explicit mechanism. An empty value (null) selects the server's
+ * default mechanism and is always allowed, so it is not checked here.
+ */
+ public static void validateS3CredentialVendingMechanismAllowed(
+ @NonNull RealmConfig realmConfig, @Nullable String mechanism) {
+ if (mechanism == null) {
+ return;
+ }
+ List<String> allowed = realmConfig.getConfig(SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS);
+ if (!allowed.contains(mechanism)) {
+ throw new ValidationException(
+ "S3 credential vending mechanism %s is not enabled in this realm", mechanism);
+ }
+ }
+
public static boolean safeStorageType(String name) {
return StorageTypeFileIO.valueOf(name).safe();
}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/config/ProductionReadinessChecks.java b/runtime/service/src/main/java/org/apache/polaris/service/config/ProductionReadinessChecks.java
index 35429f6db9..c3c0908d64 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/config/ProductionReadinessChecks.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/config/ProductionReadinessChecks.java
@@ -36,6 +36,7 @@ import org.apache.polaris.core.config.ProductionReadinessCheck;
import org.apache.polaris.core.config.ProductionReadinessCheck.Error;
import org.apache.polaris.core.credentials.connection.ConnectionCredentialVendor;
import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
import org.apache.polaris.service.auth.AuthenticationConfiguration;
import org.apache.polaris.service.auth.AuthenticationRealmConfiguration.TokenBrokerConfiguration.RSAKeyPairConfiguration;
import org.apache.polaris.service.auth.AuthenticationRealmConfiguration.TokenBrokerConfiguration.SymmetricKeyConfiguration;
@@ -49,6 +50,7 @@ import org.apache.polaris.service.context.TestRealmContextResolver;
import org.apache.polaris.service.credentials.connection.AuthType;
import org.apache.polaris.service.metrics.MetricsConfiguration;
import org.apache.polaris.service.persistence.InMemoryPolarisMetaStoreManagerFactory;
+import org.apache.polaris.service.storage.S3CredentialVendingMechanisms;
import org.eclipse.microprofile.config.Config;
import org.eclipse.microprofile.config.ConfigValue;
import org.slf4j.Logger;
@@ -446,6 +448,80 @@ public class ProductionReadinessChecks {
: ProductionReadinessCheck.of(errors.toArray(new Error[0]));
}
+ /**
+ * Every explicit name in {@code SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS}, in the defaults (or
+ * the flag's code default when the defaults do not set it) and in each realm override, must be
+ * installed in this server. A server without an {@code STS} mechanism is a severe issue, because
+ * every catalog that leaves {@code credentialVendingMechanism} empty selects it. A
+ * listed-but-uninstalled mechanism is not severe: the mechanism is refused at catalog create and
+ * update, and whenever a credential is vended for a catalog that selects it. Apart from a missing
+ * {@code STS}, only the registry's own constructor (a bean with no {@code @Identifier}, or two
+ * beans sharing one) aborts startup.
+ */
+ @Produces
+ public ProductionReadinessCheck checkS3CredentialVendingMechanisms(
+ FeaturesConfiguration featureConfiguration, S3CredentialVendingMechanisms mechanisms) {
+ var flag = FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS;
+ var mapper = JsonMapper.builder().build();
+ var errors = new ArrayList<Error>();
+ if (!mechanisms.isAvailable(S3CredentialVendingMechanism.STS)) {
+ errors.add(
+ Error.ofSevere(
+ format(
+ "No S3 credential vending mechanism carries the STS identifier; catalogs "
+ + "that leave credentialVendingMechanism empty cannot vend credentials. "
+ + "Available: %s",
+ mechanisms.availableIds()),
+ "S3CredentialVendingMechanism @Identifier(\"STS\")"));
+ }
+ @SuppressWarnings("unchecked")
+ var defaults =
+ (List<String>)
+ featureConfiguration
+ .parseDefaults(mapper)
+ .getOrDefault(flag.key(), flag.defaultValue());
+ defaults.forEach(
+ name ->
+ checkMechanismAvailable(
+ name, format("polaris.features.\"%s\"", flag.key()), mechanisms, errors));
+ featureConfiguration
+ .parseRealmOverrides(mapper)
+ .forEach(
+ (realmId, overrides) -> {
+ @SuppressWarnings("unchecked")
+ var names = (List<String>) overrides.getOrDefault(flag.key(), List.of());
+ names.forEach(
+ name ->
+ checkMechanismAvailable(
+ name,
+ format(
+ "polaris.features.realm-overrides.\"%s\".overrides.\"%s\"",
+ realmId, flag.key()),
+ mechanisms,
+ errors));
+ });
+ return errors.isEmpty()
+ ? ProductionReadinessCheck.OK
+ : ProductionReadinessCheck.of(errors.toArray(new Error[0]));
+ }
+
+ private static void checkMechanismAvailable(
+ String name,
+ String offendingProperty,
+ S3CredentialVendingMechanisms mechanisms,
+ List<Error> errors) {
+ if (!mechanisms.isAvailable(name)) {
+ List<String> available = new ArrayList<>(mechanisms.availableIds());
+ errors.add(
+ Error.of(
+ format(
+ "S3 credential vending mechanism '%s' listed in %s is not available in this "
+ + "server; catalogs that select it are refused at use. Available: %s",
+ name, offendingProperty, available),
+ offendingProperty));
+ }
+ }
+
@Produces
public ProductionReadinessCheck checkSkipCredentialSubscopingIndirection(
FeaturesConfiguration featureConfiguration) {
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 23c985ab5b..8ca071f564 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
@@ -29,7 +29,6 @@ import java.util.List;
import java.util.Optional;
import java.util.function.Function;
import java.util.function.Supplier;
-import org.apache.polaris.core.config.FeatureConfiguration;
import org.apache.polaris.core.config.RealmConfig;
import org.apache.polaris.core.entity.PolarisEntity;
import org.apache.polaris.core.storage.CredentialVendingContext;
@@ -38,55 +37,41 @@ 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.StorageAccessConfig;
-import org.apache.polaris.core.storage.aws.AwsCredentialsStorageIntegration;
import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
-import org.apache.polaris.core.storage.aws.StsClientProvider;
import org.apache.polaris.core.storage.azure.AzureCredentialsStorageIntegration;
import org.apache.polaris.core.storage.azure.AzureStorageConfigurationInfo;
import org.apache.polaris.core.storage.cache.StorageCredentialCache;
import org.apache.polaris.core.storage.gcp.GcpCredentialsStorageIntegration;
import org.apache.polaris.core.storage.gcp.GcpStorageConfigurationInfo;
+import org.apache.polaris.service.catalog.validation.IcebergPropertiesValidation;
import org.jspecify.annotations.NonNull;
import org.jspecify.annotations.Nullable;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
/**
* Provider that returns a {@link PolarisStorageIntegration} for a resolved entity path. A fresh
* integration is constructed per call; the per-config state inside each integration is a thin
- * wrapper over shared, application-scoped resources (STS client provider, GCP credentials supplier,
- * Google HTTP transport factory, request-scope realm config).
+ * wrapper over shared, application-scoped resources (the credential vending mechanism registry, GCP
+ * credentials supplier, Google HTTP transport factory, request-scope realm config).
*/
@ApplicationScoped
public class PolarisStorageIntegrationProviderImpl implements PolarisStorageIntegrationProvider {
- private final Function<AwsStorageConfigurationInfo, AwsCredentialsStorageIntegration> awsFactory;
+ private final S3CredentialVendingMechanisms mechanisms;
private final Function<GcpStorageConfigurationInfo, GcpCredentialsStorageIntegration> gcpFactory;
private final Function<AzureStorageConfigurationInfo, AzureCredentialsStorageIntegration>
azureFactory;
+ private final RealmConfig realmConfig;
@SuppressWarnings("CdiInjectionPointsInspection")
@Inject
public PolarisStorageIntegrationProviderImpl(
StorageConfiguration storageConfiguration,
- StsClientProvider stsClientProvider,
+ S3CredentialVendingMechanisms mechanisms,
RealmConfig realmConfig,
Clock clock,
StorageCredentialCache cache) {
- this.awsFactory =
- storageConfig ->
- new AwsCredentialsStorageIntegration(
- stsClientProvider,
- config -> {
- if (realmConfig.getConfig(
- FeatureConfiguration.RESOLVE_CREDENTIALS_BY_STORAGE_NAME)) {
- return Optional.of(
- storageConfiguration.stsCredentials(config.getStorageName()));
- }
- return Optional.of(storageConfiguration.stsCredentials());
- },
- cache,
- storageConfig,
- realmConfig);
+ this.realmConfig = realmConfig;
+ this.mechanisms = mechanisms;
Supplier<GoogleCredentials> gcpCredsProvider =
storageConfiguration.gcpCredentialsSupplier(clock);
HttpTransportFactory gcpTransportFactory =
@@ -100,15 +85,12 @@ public class PolarisStorageIntegrationProviderImpl implements PolarisStorageInte
}
public PolarisStorageIntegrationProviderImpl(
- StsClientProvider stsClientProvider,
- Optional<AwsCredentialsProvider> stsCredentials,
+ S3CredentialVendingMechanisms mechanisms,
Supplier<GoogleCredentials> gcpCredsProvider,
StorageCredentialCache cache,
RealmConfig realmConfig) {
- this.awsFactory =
- storageConfig ->
- new AwsCredentialsStorageIntegration(
- stsClientProvider, config -> stsCredentials, cache, storageConfig, realmConfig);
+ this.realmConfig = realmConfig;
+ this.mechanisms = mechanisms;
HttpTransportFactory gcpTransportFactory =
ServiceOptions.getFromServiceLoader(HttpTransportFactory.class, NetHttpTransport::new);
this.gcpFactory =
@@ -130,7 +112,16 @@ public class PolarisStorageIntegrationProviderImpl implements PolarisStorageInte
private PolarisStorageIntegration createIntegration(
PolarisStorageConfigurationInfo storageConfig) {
return switch (storageConfig.getStorageType()) {
- case S3 -> awsFactory.apply((AwsStorageConfigurationInfo) storageConfig);
+ case S3 -> {
+ AwsStorageConfigurationInfo awsConfig = (AwsStorageConfigurationInfo) storageConfig;
+ // The only check at vending time: the realm allowlist on the explicit value, then the
+ // registry. Catalog create and update ran the same two checks before the config was stored.
+ IcebergPropertiesValidation.validateS3CredentialVendingMechanismAllowed(
+ realmConfig, awsConfig.getCredentialVendingMechanism());
+ yield mechanisms
+ .require(awsConfig.resolvedCredentialVendingMechanism())
+ .integrationFor(awsConfig);
+ }
case GCS -> gcpFactory.apply((GcpStorageConfigurationInfo) storageConfig);
case AZURE -> azureFactory.apply((AzureStorageConfigurationInfo) storageConfig);
case FILE -> FILE_INTEGRATION;
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/storage/S3CredentialVendingMechanisms.java b/runtime/service/src/main/java/org/apache/polaris/service/storage/S3CredentialVendingMechanisms.java
new file mode 100644
index 0000000000..8cb04689ff
--- /dev/null
+++ b/runtime/service/src/main/java/org/apache/polaris/service/storage/S3CredentialVendingMechanisms.java
@@ -0,0 +1,127 @@
+/*
+ * 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;
+
+import io.quarkus.arc.ClientProxy;
+import io.quarkus.runtime.Startup;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.enterprise.inject.spi.Bean;
+import jakarta.enterprise.inject.spi.BeanManager;
+import jakarta.inject.Inject;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
+
+/**
+ * Every {@link S3CredentialVendingMechanism} installed in this server, resolved once at startup.
+ * The identifiers come from the {@code @Identifier} qualifier of every enabled bean, read through
+ * the {@link BeanManager} so that an enabled alternative does not hide the bean it replaces; each
+ * identifier is then resolved through {@link Instance#select}, which lets the alternative with the
+ * highest priority win, and the instance is created so a bean that cannot be built fails startup
+ * instead of the first request. A realm's allowlist is checked separately; this registry answers
+ * only what is installed.
+ */
+@ApplicationScoped
+@Startup
+public class S3CredentialVendingMechanisms {
+
+ private final Map<String, S3CredentialVendingMechanism> mechanisms;
+
+ @Inject
+ public S3CredentialVendingMechanisms(
+ @Any Instance<S3CredentialVendingMechanism> candidates, BeanManager beanManager) {
+ // getBeans(Type, Annotation...) returns every enabled bean without ambiguity resolution.
+ // Instance#handles() on the unqualified @Any Instance would resolve ambiguity across all
+ // mechanism beans, which drops the bean an enabled alternative replaces.
+ Set<String> ids = new TreeSet<>();
+ for (Bean<?> bean :
+ beanManager.getBeans(S3CredentialVendingMechanism.class, Any.Literal.INSTANCE)) {
+ ids.add(identifierOf(bean));
+ }
+ Map<String, S3CredentialVendingMechanism> resolved = new TreeMap<>();
+ for (String id : ids) {
+ Instance<S3CredentialVendingMechanism> selected =
+ candidates.select(Identifier.Literal.of(id));
+ if (selected.isAmbiguous()) {
+ throw new IllegalStateException(
+ "Two S3 credential vending mechanisms share the identifier " + id);
+ }
+ if (selected.isUnsatisfied()) {
+ throw new IllegalStateException(
+ "S3 credential vending mechanism " + id + " resolves to no enabled bean");
+ }
+ S3CredentialVendingMechanism mechanism = selected.get();
+ // Normal-scoped beans hand out a client proxy; unwrapping it creates the instance now, so a
+ // constructor or injection failure aborts startup rather than the first request.
+ ClientProxy.unwrap(mechanism);
+ resolved.put(id, mechanism);
+ }
+ this.mechanisms = Collections.unmodifiableMap(resolved);
+ }
+
+ /**
+ * For tests: a live view of identifier to mechanism. The map is read on every call, so a test can
+ * install or remove a mechanism after catalogs were created.
+ */
+ public S3CredentialVendingMechanisms(Map<String, S3CredentialVendingMechanism> mechanisms) {
+ this.mechanisms = mechanisms;
+ }
+
+ /** The installed identifiers, sorted. */
+ public Set<String> availableIds() {
+ return Collections.unmodifiableSet(new TreeSet<>(mechanisms.keySet()));
+ }
+
+ public boolean isAvailable(String id) {
+ return mechanisms.containsKey(id);
+ }
+
+ /**
+ * The mechanism for an identifier. A missing identifier is an {@link IllegalArgumentException},
+ * which the exception mapper turns into the same 400 that catalog create, catalog update and
+ * credential vending return for a mechanism this server lacks.
+ */
+ public S3CredentialVendingMechanism require(String id) {
+ S3CredentialVendingMechanism mechanism = mechanisms.get(id);
+ if (mechanism == null) {
+ throw new IllegalArgumentException(
+ String.format("S3 credential vending mechanism %s is not available in this server", id));
+ }
+ return mechanism;
+ }
+
+ private static String identifierOf(Bean<?> bean) {
+ return bean.getQualifiers().stream()
+ .filter(Identifier.class::isInstance)
+ .map(qualifier -> ((Identifier) qualifier).value())
+ .findFirst()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "S3 credential vending mechanism bean "
+ + bean.getBeanClass().getName()
+ + " has no @Identifier"));
+ }
+}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/storage/StsCredentialVendingMechanism.java b/runtime/service/src/main/java/org/apache/polaris/service/storage/StsCredentialVendingMechanism.java
new file mode 100644
index 0000000000..80f4c5b7d0
--- /dev/null
+++ b/runtime/service/src/main/java/org/apache/polaris/service/storage/StsCredentialVendingMechanism.java
@@ -0,0 +1,97 @@
+/*
+ * 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;
+
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.Optional;
+import java.util.function.Function;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.storage.PolarisStorageIntegration;
+import org.apache.polaris.core.storage.aws.AwsCredentialsStorageIntegration;
+import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
+import org.apache.polaris.core.storage.aws.StsClientProvider;
+import org.apache.polaris.core.storage.cache.StorageCredentialCache;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+
+/**
+ * STS AssumeRole against the catalog's role. A catalog that leaves {@code
+ * credentialVendingMechanism} empty uses this mechanism. A server build replaces it with an
+ * {@code @Alternative} bean of a higher {@code @Priority} that carries {@code @Identifier("STS")}.
+ */
+@ApplicationScoped
+@Identifier(S3CredentialVendingMechanism.STS)
+public class StsCredentialVendingMechanism implements S3CredentialVendingMechanism {
+
+ private final StsClientProvider stsClientProvider;
+ private final Function<AwsStorageConfigurationInfo, Optional<AwsCredentialsProvider>>
+ credentialsResolver;
+ private final StorageCredentialCache cache;
+ private final RealmConfig realmConfig;
+
+ /**
+ * For the client proxy ArC generates for this normal-scoped bean: the proxy subclass needs a
+ * no-args constructor. A working instance always comes from one of the constructors below.
+ */
+ protected StsCredentialVendingMechanism() {
+ this.stsClientProvider = null;
+ this.credentialsResolver = null;
+ this.cache = null;
+ this.realmConfig = null;
+ }
+
+ @Inject
+ public StsCredentialVendingMechanism(
+ StorageConfiguration storageConfiguration,
+ StsClientProvider stsClientProvider,
+ StorageCredentialCache cache,
+ RealmConfig realmConfig) {
+ this.stsClientProvider = stsClientProvider;
+ this.cache = cache;
+ this.realmConfig = realmConfig;
+ this.credentialsResolver =
+ config -> {
+ if (realmConfig.getConfig(FeatureConfiguration.RESOLVE_CREDENTIALS_BY_STORAGE_NAME)) {
+ return Optional.of(storageConfiguration.stsCredentials(config.getStorageName()));
+ }
+ return Optional.of(storageConfiguration.stsCredentials());
+ };
+ }
+
+ /** Test constructor: a fixed credentials provider and the realm config to vend under. */
+ public StsCredentialVendingMechanism(
+ StsClientProvider stsClientProvider,
+ Optional<AwsCredentialsProvider> stsCredentials,
+ StorageCredentialCache cache,
+ RealmConfig realmConfig) {
+ this.stsClientProvider = stsClientProvider;
+ this.cache = cache;
+ this.realmConfig = realmConfig;
+ this.credentialsResolver = config -> stsCredentials;
+ }
+
+ @Override
+ public PolarisStorageIntegration integrationFor(AwsStorageConfigurationInfo storageConfig) {
+ return new AwsCredentialsStorageIntegration(
+ stsClientProvider, credentialsResolver, cache, storageConfig, realmConfig);
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java
index df382b3ea8..fde669bd08 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java
@@ -23,6 +23,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
import jakarta.ws.rs.core.Response;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -58,6 +59,9 @@ import org.apache.polaris.core.persistence.dao.entity.BaseResult;
import org.apache.polaris.core.persistence.dao.entity.CreateCatalogResult;
import org.apache.polaris.core.persistence.dao.entity.EntityResult;
import org.apache.polaris.core.secrets.UnsafeInMemorySecretsManager;
+import org.apache.polaris.core.storage.PolarisStorageIntegration;
+import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
import org.apache.polaris.service.TestServices;
import org.apache.polaris.service.config.ReservedProperties;
import org.apache.polaris.service.identity.provider.DefaultServiceIdentityProvider;
@@ -392,7 +396,8 @@ public class ManagementServiceTest {
new DefaultServiceIdentityProvider(),
principal,
new PolarisAuthorizerImpl(services.realmConfig()),
- ReservedProperties.NONE);
+ ReservedProperties.NONE,
+ services.vendingMechanisms());
}
private PrincipalEntity createPrincipal(
@@ -753,4 +758,481 @@ public class ManagementServiceTest {
resultWithError.getReturnStatus(),
resultWithError.getExtraInformation()));
}
+
+ @Test
+ public void anEmptyMechanismIsReadBackAsEmpty() {
+ AwsStorageConfigInfo awsConfigModel =
+ AwsStorageConfigInfo.builder()
+ .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+ .setStorageType(StorageConfigInfo.StorageTypeEnum.S3)
+ .setAllowedLocations(List.of("s3://bucket/path/to/data"))
+ .build();
+ Catalog catalog =
+ PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName("mechanism-default")
+ .setProperties(new CatalogProperties("s3://bucket/path/to/data"))
+ .setStorageConfigInfo(awsConfigModel)
+ .build();
+ try (Response response =
+ services
+ .catalogsApi()
+ .createCatalog(
+ new CreateCatalogRequest(catalog),
+ services.realmContext(),
+ services.securityContext())) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ try (Response response =
+ services
+ .catalogsApi()
+ .getCatalog("mechanism-default", services.realmContext(), services.securityContext())) {
+ Catalog fetched = (Catalog) response.getEntity();
+ assertThat(
+ ((AwsStorageConfigInfo) fetched.getStorageConfigInfo())
+ .getCredentialVendingMechanism())
+ .isNull();
+ }
+ }
+
+ private static final String TEST_MECHANISM = "TEST_MECHANISM";
+
+ private static TestServices mechanismServices(
+ List<String> mechanisms, boolean unrestrictedChanges) {
+ return TestServices.builder()
+ .config(
+ Map.of(
+ "SUPPORTED_CATALOG_STORAGE_TYPES",
+ List.of("S3", "GCS", "AZURE"),
+ "SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS",
+ mechanisms,
+ "ALLOW_UNRESTRICTED_STORAGE_CONFIG_ROLE_CHANGES",
+ unrestrictedChanges))
+ .additionalVendingMechanisms(Map.of(TEST_MECHANISM, TestServices.fakeMechanism()))
+ .build();
+ }
+
+ private static AwsStorageConfigInfo.Builder secondMechanismConfig() {
+ return AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(TEST_MECHANISM)
+ .setAllowedLocations(List.of("s3://second-bucket/base/"));
+ }
+
+ private static AwsStorageConfigInfo emptyMechanismConfig() {
+ return AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .build();
+ }
+
+ private static Catalog catalogNamed(String name, StorageConfigInfo storageConfig) {
+ return PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(name)
+ .setProperties(new CatalogProperties("s3://second-bucket/base/" + name))
+ .setStorageConfigInfo(storageConfig)
+ .build();
+ }
+
+ private static Response create(TestServices svc, Catalog catalog) {
+ return svc.catalogsApi()
+ .createCatalog(
+ new CreateCatalogRequest(catalog), svc.realmContext(), svc.securityContext());
+ }
+
+ private static Catalog fetch(TestServices svc, String name) {
+ try (Response response =
+ svc.catalogsApi().getCatalog(name, svc.realmContext(), svc.securityContext())) {
+ return (Catalog) response.getEntity();
+ }
+ }
+
+ @Test
+ public void testASecondMechanismIsRejectedByTheDefaultAllowlist() {
+ TestServices defaults = mechanismServices(List.of("STS"), false);
+ assertThatThrownBy(
+ () -> create(defaults, catalogNamed("second-off", secondMechanismConfig().build())))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(
+ "S3 credential vending mechanism " + TEST_MECHANISM + " is not enabled in this realm");
+ }
+
+ @Test
+ public void anExplicitStsIsRejectedWhenTheRealmListsOnlyASecondMechanism() {
+ TestServices secondOnly = mechanismServices(List.of(TEST_MECHANISM), false);
+ AwsStorageConfigInfo sts =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .setCredentialVendingMechanism("STS")
+ .build();
+ assertThatThrownBy(() -> create(secondOnly, catalogNamed("sts-off", sts)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage("S3 credential vending mechanism STS is not enabled in this realm");
+ }
+
+ @Test
+ public void anEmptyMechanismIsAcceptedWhenTheRealmListsOnlyASecondMechanism() {
+ TestServices secondOnly = mechanismServices(List.of(TEST_MECHANISM), false);
+ AwsStorageConfigInfo empty =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .build();
+ try (Response response = create(secondOnly, catalogNamed("empty-allowed", empty))) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ }
+
+ @Test
+ public void testDisallowedMechanismIsRejectedOnUpdateToo() {
+ TestServices stsOnlyUnrestricted = mechanismServices(List.of("STS"), true);
+ AwsStorageConfigInfo emptyMechanism =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .build();
+ try (Response response =
+ create(stsOnlyUnrestricted, catalogNamed("empty-stay", emptyMechanism))) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ Catalog fetched = fetch(stsOnlyUnrestricted, "empty-stay");
+ UpdateCatalogRequest toSecond =
+ new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of("default-base-location", "s3://second-bucket/base/empty-stay"),
+ secondMechanismConfig().build());
+ assertThatThrownBy(
+ () ->
+ stsOnlyUnrestricted
+ .catalogsApi()
+ .updateCatalog(
+ "empty-stay",
+ toSecond,
+ stsOnlyUnrestricted.realmContext(),
+ stsOnlyUnrestricted.securityContext()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(
+ "S3 credential vending mechanism " + TEST_MECHANISM + " is not enabled in this realm");
+ }
+
+ @Test
+ public void theLiteralDefaultIsRefusedLikeAnyUnknownValueAtCreateAndUpdate() {
+ TestServices svc = mechanismServices(List.of("STS"), true);
+ String refused = "S3 credential vending mechanism DEFAULT is not enabled in this realm";
+ AwsStorageConfigInfo namesDefault =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism("DEFAULT")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .build();
+ assertThatThrownBy(() -> create(svc, catalogNamed("names-default", namesDefault)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(refused);
+
+ AwsStorageConfigInfo empty =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .build();
+ try (Response response = create(svc, catalogNamed("stays-empty", empty))) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ Catalog fetched = fetch(svc, "stays-empty");
+ UpdateCatalogRequest toEmpty =
+ new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of("default-base-location", "s3://second-bucket/base/stays-empty"),
+ namesDefault);
+ assertThatThrownBy(
+ () ->
+ svc.catalogsApi()
+ .updateCatalog(
+ "stays-empty", toEmpty, svc.realmContext(), svc.securityContext()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(refused);
+ }
+
+ @Test
+ public void anEmptyMechanismCanBeUpdatedToAnExplicitSts() {
+ TestServices svc = mechanismServices(List.of("STS"), false);
+ AwsStorageConfigInfo empty =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .build();
+ AwsStorageConfigInfo explicitSts =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism("STS")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .build();
+ try (Response response = create(svc, catalogNamed("empty-then-sts", empty))) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ Catalog fetched = fetch(svc, "empty-then-sts");
+ UpdateCatalogRequest toSts =
+ new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of("default-base-location", "s3://second-bucket/base/empty-then-sts"),
+ explicitSts);
+ try (Response response =
+ svc.catalogsApi()
+ .updateCatalog("empty-then-sts", toSts, svc.realmContext(), svc.securityContext())) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ assertThat(
+ ((AwsStorageConfigInfo) fetch(svc, "empty-then-sts").getStorageConfigInfo())
+ .getCredentialVendingMechanism())
+ .isEqualTo("STS");
+ }
+
+ @Test
+ public void anExplicitStsCanBeUpdatedBackToTheEmptyField() {
+ EndpointRequiringMechanism recording = new EndpointRequiringMechanism();
+ TestServices svc =
+ TestServices.builder()
+ .config(
+ Map.of(
+ "SUPPORTED_CATALOG_STORAGE_TYPES", List.of("S3"),
+ "SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS", List.of("STS")))
+ .additionalVendingMechanisms(Map.of(S3CredentialVendingMechanism.STS, recording))
+ .build();
+ AwsStorageConfigInfo explicitSts =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism("STS")
+ .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+ .setExternalId("external-id")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .setEndpoint("https://s3.example.test")
+ .build();
+ try (Response response = create(svc, catalogNamed("sts-then-empty", explicitSts))) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ Catalog fetched = fetch(svc, "sts-then-empty");
+ AwsStorageConfigInfo backToDefault =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+ .setExternalId("external-id")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .setEndpoint("https://s3.example.test")
+ .build();
+ UpdateCatalogRequest toEmpty =
+ new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of("default-base-location", "s3://second-bucket/base/sts-then-empty"),
+ backToDefault);
+ try (Response response =
+ svc.catalogsApi()
+ .updateCatalog("sts-then-empty", toEmpty, svc.realmContext(), svc.securityContext())) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ assertThat(
+ ((AwsStorageConfigInfo) fetch(svc, "sts-then-empty").getStorageConfigInfo())
+ .getCredentialVendingMechanism())
+ .isNull();
+ // The STS override validates the explicit create, then the update back to the empty field.
+ assertThat(recording.currents).hasSize(2);
+ assertThat(recording.currents.get(0)).isNull();
+ assertThat(recording.currents.get(1).getCredentialVendingMechanism()).isEqualTo("STS");
+ assertThat(recording.updateds).hasSize(2);
+ assertThat(recording.updateds.get(0).getCredentialVendingMechanism()).isEqualTo("STS");
+ assertThat(recording.updateds.get(1).getCredentialVendingMechanism()).isNull();
+ }
+
+ @Test
+ public void testASecondMechanismCatalogIsCreatedAndReadBack() {
+ TestServices enabled = mechanismServices(List.of("STS", TEST_MECHANISM), false);
+ try (Response response =
+ create(enabled, catalogNamed("second-on", secondMechanismConfig().build()))) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ AwsStorageConfigInfo fetched =
+ (AwsStorageConfigInfo) fetch(enabled, "second-on").getStorageConfigInfo();
+ assertThat(fetched.getCredentialVendingMechanism()).isEqualTo(TEST_MECHANISM);
+ }
+
+ @Test
+ public void changingTheMechanismIsAcceptedAndValidatedByTheNewMechanism() {
+ EndpointRequiringMechanism mechanism = new EndpointRequiringMechanism();
+ TestServices svc =
+ TestServices.builder()
+ .config(
+ Map.of(
+ "SUPPORTED_CATALOG_STORAGE_TYPES", List.of("S3"),
+ "SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS", List.of("STS", TEST_MECHANISM)))
+ .additionalVendingMechanisms(Map.of(TEST_MECHANISM, mechanism))
+ .build();
+ AwsStorageConfigInfo sts =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism("STS")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .build();
+ try (Response response = create(svc, catalogNamed("mechanism-change", sts))) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ Catalog fetched = fetch(svc, "mechanism-change");
+ UpdateCatalogRequest toTestMechanism =
+ new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of("default-base-location", "s3://second-bucket/base/mechanism-change"),
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(TEST_MECHANISM)
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .setEndpoint("https://s3.example.test")
+ .build());
+ try (Response response =
+ svc.catalogsApi()
+ .updateCatalog(
+ "mechanism-change", toTestMechanism, svc.realmContext(), svc.securityContext())) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ assertThat(
+ ((AwsStorageConfigInfo) fetch(svc, "mechanism-change").getStorageConfigInfo())
+ .getCredentialVendingMechanism())
+ .isEqualTo(TEST_MECHANISM);
+ assertThat(mechanism.currents).hasSize(1);
+ assertThat(mechanism.currents.get(0).getCredentialVendingMechanism()).isEqualTo("STS");
+ assertThat(mechanism.updateds).hasSize(1);
+ assertThat(mechanism.updateds.get(0).getCredentialVendingMechanism()).isEqualTo(TEST_MECHANISM);
+ }
+
+ @Test
+ public void emptyMechanismCatalogEndpointStaysMutable() {
+ // An endpoint change on a catalog with an empty mechanism is accepted and read back.
+ TestServices svc = mechanismServices(List.of("STS"), false);
+ AwsStorageConfigInfo emptyMechanism =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .setEndpoint("https://s3.example.com:1234")
+ .setPathStyleAccess(true)
+ .build();
+ try (Response response = create(svc, catalogNamed("empty-mutable", emptyMechanism))) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ Catalog fetched = fetch(svc, "empty-mutable");
+ UpdateCatalogRequest updateEndpoint =
+ new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of("default-base-location", "s3://second-bucket/base/empty-mutable"),
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .setEndpoint("https://s3.other.example.com:1234")
+ .setPathStyleAccess(true)
+ .build());
+ try (Response response =
+ svc.catalogsApi()
+ .updateCatalog(
+ "empty-mutable", updateEndpoint, svc.realmContext(), svc.securityContext())) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ AwsStorageConfigInfo updated =
+ (AwsStorageConfigInfo) fetch(svc, "empty-mutable").getStorageConfigInfo();
+ assertThat(updated.getEndpoint()).isEqualTo("https://s3.other.example.com:1234");
+ }
+
+ @Test
+ public void creatingACatalogWithAnAllowlistedButUninstalledMechanismIsRefused() {
+ TestServices svc = mechanismServices(List.of("STS", "UNINSTALLED_MECHANISM"), false);
+ AwsStorageConfigInfo uninstalled =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism("UNINSTALLED_MECHANISM")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .build();
+ assertThatThrownBy(() -> create(svc, catalogNamed("never-usable", uninstalled)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "S3 credential vending mechanism UNINSTALLED_MECHANISM is not available in this server");
+ }
+
+ @Test
+ public void updatingACatalogToAnAllowlistedButUninstalledMechanismIsRefused() {
+ TestServices svc = mechanismServices(List.of("STS", "UNINSTALLED_MECHANISM"), true);
+ try (Response response = create(svc, catalogNamed("empty-stays", emptyMechanismConfig()))) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ Catalog fetched = fetch(svc, "empty-stays");
+ UpdateCatalogRequest toUninstalled =
+ new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of("default-base-location", "s3://second-bucket/base/empty-stays"),
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism("UNINSTALLED_MECHANISM")
+ .setAllowedLocations(List.of("s3://second-bucket/base/"))
+ .build());
+ assertThatThrownBy(
+ () ->
+ svc.catalogsApi()
+ .updateCatalog(
+ "empty-stays", toUninstalled, svc.realmContext(), svc.securityContext()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "S3 credential vending mechanism UNINSTALLED_MECHANISM is not available in this server");
+ }
+
+ /** A mechanism that refuses any config without an endpoint, and records what it was given. */
+ private static final class EndpointRequiringMechanism implements S3CredentialVendingMechanism {
+ final List<AwsStorageConfigurationInfo> currents = new ArrayList<>();
+ final List<AwsStorageConfigurationInfo> updateds = new ArrayList<>();
+
+ @Override
+ public PolarisStorageIntegration integrationFor(AwsStorageConfigurationInfo storageConfig) {
+ return TestServices.fakeMechanism().integrationFor(storageConfig);
+ }
+
+ @Override
+ public void validate(AwsStorageConfigurationInfo current, AwsStorageConfigurationInfo updated) {
+ currents.add(current);
+ updateds.add(updated);
+ if (updated.getEndpoint() == null) {
+ throw new IllegalArgumentException("endpoint is required for the TEST_MECHANISM mechanism");
+ }
+ }
+ }
+
+ @Test
+ public void theMechanismValidatesTheConfigAtCreateAndAtUpdate() {
+ EndpointRequiringMechanism mechanism = new EndpointRequiringMechanism();
+ TestServices svc =
+ TestServices.builder()
+ .config(
+ Map.of(
+ "SUPPORTED_CATALOG_STORAGE_TYPES", List.of("S3"),
+ "SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS", List.of("STS", TEST_MECHANISM)))
+ .additionalVendingMechanisms(Map.of(TEST_MECHANISM, mechanism))
+ .build();
+ AwsStorageConfigInfo.Builder base =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(TEST_MECHANISM)
+ .setAllowedLocations(List.of("s3://second-bucket/base/"));
+
+ assertThatThrownBy(() -> create(svc, catalogNamed("needs-endpoint", base.build())))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("endpoint is required for the TEST_MECHANISM mechanism");
+ assertThat(mechanism.currents).containsExactly((AwsStorageConfigurationInfo) null);
+
+ try (Response response =
+ create(
+ svc,
+ catalogNamed("needs-endpoint", base.setEndpoint("https://s3.example.test").build()))) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ assertThat(mechanism.currents).hasSize(2);
+ assertThat(mechanism.updateds.get(1).getEndpoint()).isEqualTo("https://s3.example.test");
+
+ Catalog fetched = fetch(svc, "needs-endpoint");
+ UpdateCatalogRequest dropEndpoint =
+ new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of("default-base-location", "s3://second-bucket/base/needs-endpoint"),
+ base.setEndpoint(null).build());
+ assertThatThrownBy(
+ () ->
+ svc.catalogsApi()
+ .updateCatalog(
+ "needs-endpoint", dropEndpoint, svc.realmContext(), svc.securityContext()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("endpoint is required for the TEST_MECHANISM mechanism");
+ assertThat(mechanism.currents).hasSize(3);
+ assertThat(mechanism.currents.get(2)).isNotNull();
+ assertThat(mechanism.currents.get(2).getEndpoint()).isEqualTo("https://s3.example.test");
+ }
}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceAuthzTest.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceAuthzTest.java
index 9cf3128777..f068426f58 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceAuthzTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceAuthzTest.java
@@ -40,14 +40,22 @@ import org.apache.polaris.core.entity.CatalogRoleEntity;
import org.apache.polaris.core.entity.PolarisPrivilege;
import org.apache.polaris.core.entity.PrincipalEntity;
import org.apache.polaris.core.entity.PrincipalRoleEntity;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
import org.apache.polaris.service.Profiles;
+import org.apache.polaris.service.storage.S3CredentialVendingMechanisms;
import org.junit.jupiter.api.DynamicNode;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestFactory;
+import org.mockito.Mockito;
@QuarkusTest
@TestProfile(Profiles.PolarisAuthzBaseProfile.class)
public class PolarisAdminServiceAuthzTest extends PolarisAuthzTestBase {
+ private final S3CredentialVendingMechanism stsMechanism =
+ Mockito.mock(S3CredentialVendingMechanism.class);
+ private final S3CredentialVendingMechanisms vendingMechanisms =
+ new S3CredentialVendingMechanisms(Map.of("STS", stsMechanism));
+
private PolarisAdminService newTestAdminService() {
final PolarisPrincipal authenticatedPrincipal =
PolarisPrincipal.of(
@@ -65,7 +73,8 @@ public class PolarisAdminServiceAuthzTest extends PolarisAuthzTestBase {
serviceIdentityProvider,
authenticatedPrincipal,
polarisAuthorizer,
- reservedProperties);
+ reservedProperties,
+ vendingMechanisms);
}
private PolarisAdminService newExternalTestAdminService(String principalName) {
@@ -84,7 +93,8 @@ public class PolarisAdminServiceAuthzTest extends PolarisAuthzTestBase {
serviceIdentityProvider,
externalPrincipal,
polarisAuthorizer,
- reservedProperties);
+ reservedProperties,
+ vendingMechanisms);
}
private PolarisAdminService newTestAdminService(Set<String> activatedPrincipalRoles) {
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceTest.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceTest.java
index d0b950a090..ada1d9a50d 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceTest.java
@@ -38,8 +38,10 @@ import java.util.Set;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.ForbiddenException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.exceptions.ValidationException;
import org.apache.polaris.core.PolarisCallContext;
import org.apache.polaris.core.admin.model.AuthenticationParameters;
import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
@@ -51,6 +53,7 @@ import org.apache.polaris.core.admin.model.CreateCatalogRequest;
import org.apache.polaris.core.admin.model.ExternalCatalog;
import org.apache.polaris.core.admin.model.IcebergRestConnectionConfigInfo;
import org.apache.polaris.core.admin.model.OAuthClientCredentialsParameters;
+import org.apache.polaris.core.admin.model.PolarisCatalog;
import org.apache.polaris.core.admin.model.StorageConfigInfo;
import org.apache.polaris.core.auth.AuthorizationDecision;
import org.apache.polaris.core.auth.AuthorizationState;
@@ -95,7 +98,9 @@ import org.apache.polaris.core.persistence.resolver.ResolvedPathKey;
import org.apache.polaris.core.persistence.resolver.ResolverStatus;
import org.apache.polaris.core.secrets.SecretReference;
import org.apache.polaris.core.secrets.UserSecretsManager;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
import org.apache.polaris.service.config.ReservedProperties;
+import org.apache.polaris.service.storage.S3CredentialVendingMechanisms;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -116,6 +121,7 @@ public class PolarisAdminServiceTest {
@Mock private PolarisResolutionManifest resolutionManifest;
@Mock private PolarisResolvedPathWrapper resolvedPathWrapper;
@Mock private RealmConfig realmConfig;
+ @Mock private S3CredentialVendingMechanism stsMechanism;
private PolarisAdminService adminService;
@@ -151,6 +157,8 @@ public class PolarisAdminServiceTest {
authenticatedPrincipal = PolarisPrincipal.of("test", AttributeMap.EMPTY, Set.of());
+ S3CredentialVendingMechanisms vendingMechanisms =
+ new S3CredentialVendingMechanisms(Map.of("STS", stsMechanism));
adminService =
new PolarisAdminService(
callContext,
@@ -160,7 +168,8 @@ public class PolarisAdminServiceTest {
identityProvider,
authenticatedPrincipal,
authorizer,
- reservedProperties);
+ reservedProperties,
+ vendingMechanisms);
}
protected static void assertSuccess(BaseResult result) {
@@ -240,6 +249,78 @@ public class PolarisAdminServiceTest {
verify(userSecretsManager, never()).deleteSecret(any());
}
+ private static CreateCatalogRequest secondMechanismCatalogRequest() {
+ return mechanismCatalogRequest("SECOND_MECHANISM");
+ }
+
+ private static CreateCatalogRequest stsCatalogRequest() {
+ return mechanismCatalogRequest("STS");
+ }
+
+ private static CreateCatalogRequest mechanismCatalogRequest(String mechanism) {
+ AwsStorageConfigInfo storage =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(mechanism)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/"))
+ .build();
+ return new CreateCatalogRequest(
+ PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName("second")
+ .setProperties(new CatalogProperties("s3://bucket/base/"))
+ .setStorageConfigInfo(storage)
+ .build());
+ }
+
+ /** A denied caller gets the 403 and the realm's allowlist is never consulted. */
+ @Test
+ void deniedCreateCatalogNeverConsultsTheAllowlist() {
+ when(authorizer.authorize(any(), any())).thenReturn(AuthorizationDecision.deny("denied"));
+ assertThatThrownBy(() -> adminService.createCatalog(secondMechanismCatalogRequest()))
+ .isInstanceOf(ForbiddenException.class)
+ .hasMessage("denied");
+ verify(realmConfig, never())
+ .getConfig(FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS);
+ verify(metaStoreManager, never()).createCatalog(any(), any(), any());
+ }
+
+ /** The check still runs: an authorized caller with a disabled mechanism gets the 400. */
+ @Test
+ void authorizedCreateCatalogWithADisabledMechanismIsRefusedAfterAuthorization() {
+ when(realmConfig.getConfig(FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS))
+ .thenReturn(List.of("STS"));
+ assertThatThrownBy(() -> adminService.createCatalog(secondMechanismCatalogRequest()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(
+ "S3 credential vending mechanism SECOND_MECHANISM is not enabled in this realm");
+ verify(authorizer).authorize(any(), any());
+ verify(metaStoreManager, never()).createCatalog(any(), any(), any());
+ }
+
+ /** The registry this test's adminService holds carries only an STS mock. */
+ @Test
+ void authorizedCreateCatalogWithAnUninstalledMechanismIsRefusedAfterAuthorization() {
+ when(realmConfig.getConfig(FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS))
+ .thenReturn(List.of("STS", "SECOND_MECHANISM"));
+ assertThatThrownBy(() -> adminService.createCatalog(secondMechanismCatalogRequest()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "S3 credential vending mechanism SECOND_MECHANISM is not available in this server");
+ }
+
+ /**
+ * A denied caller gets the 403 before the mechanism the registry does hold, STS, is ever
+ * validated.
+ */
+ @Test
+ void deniedCreateCatalogNeverValidatesTheMechanism() {
+ when(authorizer.authorize(any(), any())).thenReturn(AuthorizationDecision.deny("denied"));
+ assertThatThrownBy(() -> adminService.createCatalog(stsCatalogRequest()))
+ .isInstanceOf(ForbiddenException.class);
+ verify(stsMechanism, never()).validate(any(), any());
+ }
+
@Test
void testDeleteCatalogDeletesInlineOauthSecret() {
SecretReference secretReference =
@@ -993,6 +1074,10 @@ public class PolarisAdminServiceTest {
when(realmConfig.getConfig(
FeatureConfiguration.SUPPORTED_EXTERNAL_CATALOG_AUTHENTICATION_TYPES))
.thenReturn(List.of(authType.name()));
+ // The admin service now reads the realm's S3 credential vending mechanism allowlist on
+ // catalog create; the mock RealmConfig must answer it, even though these catalogs are not S3.
+ when(realmConfig.getConfig(FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS))
+ .thenReturn(List.of("STS"));
GenerateEntityIdResult idResult = mock(GenerateEntityIdResult.class);
when(idResult.getId()).thenReturn(2L);
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceTestSupport.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceTestSupport.java
index 901716760f..0108902fe9 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceTestSupport.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceTestSupport.java
@@ -18,14 +18,22 @@
*/
package org.apache.polaris.service.admin;
+import java.util.Map;
+import java.util.Optional;
import org.apache.polaris.core.auth.PolarisAuthorizer;
import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.config.RealmConfig;
import org.apache.polaris.core.context.CallContext;
import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
import org.apache.polaris.core.persistence.resolver.ResolutionManifestFactory;
import org.apache.polaris.core.secrets.UserSecretsManager;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
import org.apache.polaris.service.config.ReservedProperties;
+import org.apache.polaris.service.storage.S3CredentialVendingMechanisms;
+import org.apache.polaris.service.storage.StsCredentialVendingMechanism;
+import org.mockito.Mockito;
+import software.amazon.awssdk.services.sts.StsClient;
public final class PolarisAdminServiceTestSupport {
private PolarisAdminServiceTestSupport() {}
@@ -39,6 +47,16 @@ public final class PolarisAdminServiceTestSupport {
PolarisPrincipal principal,
PolarisAuthorizer authorizer,
ReservedProperties reservedProperties) {
+ RealmConfig realmConfig = callContext.getRealmConfig();
+ S3CredentialVendingMechanisms vendingMechanisms =
+ new S3CredentialVendingMechanisms(
+ Map.of(
+ S3CredentialVendingMechanism.STS,
+ new StsCredentialVendingMechanism(
+ destination -> Mockito.mock(StsClient.class),
+ Optional.empty(),
+ null,
+ realmConfig)));
return new PolarisAdminService(
callContext,
resolutionManifestFactory,
@@ -47,6 +65,7 @@ public final class PolarisAdminServiceTestSupport {
serviceIdentityProvider,
principal,
authorizer,
- reservedProperties);
+ reservedProperties,
+ vendingMechanisms);
}
}
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 62db133228..db0cc28c5d 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
@@ -88,6 +88,8 @@ import org.apache.polaris.service.events.PolarisEventDispatcher;
import org.apache.polaris.service.events.PolarisEventMetadataFactory;
import org.apache.polaris.service.idempotency.IdempotencyRequestContext;
import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
+import org.apache.polaris.service.storage.S3CredentialVendingMechanisms;
+import org.apache.polaris.service.storage.StsCredentialVendingMechanism;
import org.apache.polaris.service.task.TaskExecutor;
import org.apache.polaris.service.types.PolicyIdentifier;
import org.assertj.core.api.Assertions;
@@ -190,10 +192,15 @@ public abstract class PolarisAuthzTestBase {
@BeforeAll
public static void setUpMocks() {
StsClient stsClient = Mockito.mock(StsClient.class);
+ S3CredentialVendingMechanisms mechanisms =
+ new S3CredentialVendingMechanisms(
+ Map.of(
+ "STS",
+ new StsCredentialVendingMechanism(
+ destination -> stsClient, Optional.empty(), null, null)));
PolarisStorageIntegrationProviderImpl mock =
new PolarisStorageIntegrationProviderImpl(
- destination -> stsClient,
- Optional.empty(),
+ mechanisms,
() -> GoogleCredentials.create(new AccessToken("abc", new Date())),
null,
null);
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisServiceImplTest.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisServiceImplTest.java
index 2fe0e52680..18a3858a01 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisServiceImplTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisServiceImplTest.java
@@ -20,18 +20,26 @@ package org.apache.polaris.service.admin;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.lang.reflect.Method;
import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.exceptions.ForbiddenException;
import org.apache.polaris.core.admin.model.AuthenticationParameters;
+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.ConnectionConfigInfo;
+import org.apache.polaris.core.admin.model.CreateCatalogRequest;
import org.apache.polaris.core.admin.model.ExternalCatalog;
import org.apache.polaris.core.admin.model.FileStorageConfigInfo;
import org.apache.polaris.core.admin.model.PolarisCatalog;
import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.auth.AuthorizationDecision;
import org.apache.polaris.core.auth.PolarisAuthorizer;
import org.apache.polaris.core.auth.PolarisPrincipal;
import org.apache.polaris.core.config.FeatureConfiguration;
@@ -39,9 +47,12 @@ import org.apache.polaris.core.config.RealmConfig;
import org.apache.polaris.core.context.CallContext;
import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
+import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
import org.apache.polaris.core.persistence.resolver.ResolutionManifestFactory;
import org.apache.polaris.core.secrets.UserSecretsManager;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
import org.apache.polaris.service.config.ReservedProperties;
+import org.apache.polaris.service.storage.S3CredentialVendingMechanisms;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
@@ -56,6 +67,7 @@ public class PolarisServiceImplTest {
private CallContext callContext;
private ReservedProperties reservedProperties;
private RealmConfig realmConfig;
+ private S3CredentialVendingMechanism stsMechanism;
private PolarisAdminService adminService;
private PolarisServiceImpl polarisService;
@@ -70,6 +82,7 @@ public class PolarisServiceImplTest {
callContext = Mockito.mock(CallContext.class);
reservedProperties = Mockito.mock(ReservedProperties.class);
realmConfig = Mockito.mock(RealmConfig.class);
+ stsMechanism = Mockito.mock(S3CredentialVendingMechanism.class);
PolarisPrincipal principal = Mockito.mock(PolarisPrincipal.class);
when(callContext.getRealmConfig()).thenReturn(realmConfig);
@@ -79,6 +92,8 @@ public class PolarisServiceImplTest {
FeatureConfiguration.SUPPORTED_EXTERNAL_CATALOG_AUTHENTICATION_TYPES))
.thenReturn(List.of("OAUTH"));
+ S3CredentialVendingMechanisms vendingMechanisms =
+ new S3CredentialVendingMechanisms(Map.of("STS", stsMechanism));
adminService =
new PolarisAdminService(
callContext,
@@ -88,12 +103,53 @@ public class PolarisServiceImplTest {
serviceIdentityProvider,
principal,
polarisAuthorizer,
- reservedProperties);
+ reservedProperties,
+ vendingMechanisms);
polarisService =
new PolarisServiceImpl(
realmConfig, reservedProperties, adminService, serviceIdentityProvider);
}
+ /**
+ * Drives the real resource method: a denied caller gets the 403 before the realm's allowlist is
+ * read. A check re-added in front of {@code adminService.createCatalog} fails this.
+ */
+ @Test
+ void deniedCreateCatalogThroughTheResourceNeverConsultsTheAllowlist() {
+ PolarisResolutionManifest manifest = Mockito.mock(PolarisResolutionManifest.class);
+ when(resolutionManifestFactory.createResolutionManifest(any(), any())).thenReturn(manifest);
+ when(polarisAuthorizer.authorize(any(), any()))
+ .thenReturn(AuthorizationDecision.deny("denied"));
+ when(realmConfig.getConfig(FeatureConfiguration.SUPPORTED_CATALOG_STORAGE_TYPES))
+ .thenReturn(List.of("S3"));
+ when(realmConfig.getConfig(FeatureConfiguration.ALLOW_SETTING_S3_ENDPOINTS)).thenReturn(true);
+ when(realmConfig.getConfig(
+ FeatureConfiguration.ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS))
+ .thenReturn(true);
+
+ AwsStorageConfigInfo storage =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism("SECOND_MECHANISM")
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/"))
+ .build();
+ Catalog catalog =
+ PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName("second")
+ .setProperties(new CatalogProperties("s3://bucket/base/"))
+ .setStorageConfigInfo(storage)
+ .build();
+
+ assertThatThrownBy(
+ () -> polarisService.createCatalog(new CreateCatalogRequest(catalog), null, null))
+ .isInstanceOf(ForbiddenException.class)
+ .hasMessage("denied");
+ verify(realmConfig, never())
+ .getConfig(FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS);
+ verify(metaStoreManager, never()).createCatalog(any(), any(), any());
+ }
+
@Test
void testValidateExternalCatalog_InternalCatalog() {
StorageConfigInfo storageConfig =
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/S3CredentialVendingMechanismRoutesTest.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/S3CredentialVendingMechanismRoutesTest.java
new file mode 100644
index 0000000000..85816786f3
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/S3CredentialVendingMechanismRoutesTest.java
@@ -0,0 +1,360 @@
+/*
+ * 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.admin;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import jakarta.ws.rs.core.Response;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.inmemory.InMemoryFileIO;
+import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+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.CreateCatalogRequest;
+import org.apache.polaris.core.admin.model.PolarisCatalog;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.admin.model.UpdateCatalogRequest;
+import org.apache.polaris.service.TestServices;
+import org.apache.polaris.service.catalog.io.FileIOFactory;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * A stored catalog whose selected mechanism has no installed bean keeps serving metadata routes:
+ * listNamespaces, loadNamespaceMetadata and createNamespace all succeed. Only credential vending
+ * and catalog update refuse it, since the mechanism is checked where it is stored (create and
+ * update) and where it vends (the storage integration provider), never at catalog initialization.
+ * The catalog is created directly with {@link #TEST_MECHANISM}, a mechanism {@link TestServices}
+ * installs through {@code additionalVendingMechanisms}, populated while that mechanism is
+ * installed, then the "not available" scenario removes it from {@code installedMechanisms()}
+ * afterwards, which is what a server rebuilt without the mechanism looks like to a stored catalog.
+ * Every catalog gets its own allowed location: upstream rejects overlapping catalog locations at
+ * create and update.
+ */
+class S3CredentialVendingMechanismRoutesTest {
+
+ private static final String TEST_MECHANISM = "TEST_MECHANISM";
+ private static final String NOT_AVAILABLE =
+ "S3 credential vending mechanism " + TEST_MECHANISM + " is not available in this server";
+ private static final String NOT_ENABLED =
+ "S3 credential vending mechanism " + TEST_MECHANISM + " is not enabled in this realm";
+
+ /** A mutable config map: TestServices reads it live, so a test can flip the realm allowlist. */
+ private static Map<String, Object> config(boolean skipSubscoping) {
+ Map<String, Object> config = new HashMap<>();
+ config.put("SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS", List.of("STS", TEST_MECHANISM));
+ config.put("SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION", skipSubscoping);
+ return config;
+ }
+
+ private static TestServices services(Map<String, Object> config) {
+ return TestServices.builder()
+ .config(config)
+ .additionalVendingMechanisms(Map.of(TEST_MECHANISM, TestServices.fakeMechanism()))
+ .fileIOFactorySupplier(
+ () ->
+ (FileIOFactory) (accessConfig, ioImplClassName, properties) -> new InMemoryFileIO())
+ .build();
+ }
+
+ private static Catalog emptyMechanismCatalog(String name) {
+ return PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(name)
+ .setProperties(new CatalogProperties("s3://bucket/base/" + name))
+ .setStorageConfigInfo(
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/" + name + "/"))
+ .build())
+ .build();
+ }
+
+ private static void createCatalog(TestServices svc, String name) {
+ try (Response r =
+ svc.catalogsApi()
+ .createCatalog(
+ new CreateCatalogRequest(emptyMechanismCatalog(name)),
+ svc.realmContext(),
+ svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ }
+
+ private static void createNamespace(TestServices svc, String catalog, String ns) {
+ try (Response r =
+ svc.restApi()
+ .createNamespace(
+ catalog,
+ CreateNamespaceRequest.builder().withNamespace(Namespace.of(ns)).build(),
+ null,
+ svc.realmContext(),
+ svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ }
+
+ /** The table's location sits under the catalog's own allowed location. */
+ private static void createTable(TestServices svc, String catalog, String ns, String table) {
+ CreateTableRequest request =
+ CreateTableRequest.builder()
+ .withName(table)
+ .withSchema(PolarisAuthzTestBase.SCHEMA)
+ .withLocation("s3://bucket/base/" + catalog + "/" + ns + "/" + table + "/")
+ .build();
+ try (Response r =
+ svc.restApi()
+ .createTable(
+ catalog, ns, request, null, null, svc.realmContext(), svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ }
+
+ /**
+ * The allowlist ({@link ValidationException}, "not enabled") and the registry ({@link
+ * IllegalArgumentException}, "not available") refuse with different exception types, so the
+ * caller names the one the scenario reaches.
+ */
+ private static void assertLoadTableRefused(
+ TestServices svc,
+ String catalog,
+ String ns,
+ String table,
+ String accessDelegationMode,
+ Class<? extends RuntimeException> exceptionType,
+ String expectedMessage) {
+ assertThatThrownBy(
+ () ->
+ svc.restApi()
+ .loadTable(
+ catalog,
+ ns,
+ table,
+ accessDelegationMode,
+ null,
+ "ALL",
+ null,
+ svc.realmContext(),
+ svc.securityContext()))
+ .isInstanceOf(exceptionType)
+ .hasMessage(expectedMessage);
+ }
+
+ private static void assertLoadTableSucceeds(
+ TestServices svc, String catalog, String ns, String table, String accessDelegationMode) {
+ try (Response r =
+ svc.restApi()
+ .loadTable(
+ catalog,
+ ns,
+ table,
+ accessDelegationMode,
+ null,
+ "ALL",
+ null,
+ svc.realmContext(),
+ svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ }
+
+ private static void assertLoadTableSucceeds(
+ TestServices svc, String catalog, String ns, String table) {
+ assertLoadTableSucceeds(svc, catalog, ns, table, null);
+ }
+
+ private static Catalog testMechanismCatalog(String name) {
+ return PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(name)
+ .setProperties(new CatalogProperties("s3://bucket/base/" + name))
+ .setStorageConfigInfo(
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(TEST_MECHANISM)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/" + name + "/"))
+ .build())
+ .build();
+ }
+
+ private static void createTestMechanismCatalog(TestServices svc, String name) {
+ try (Response r =
+ svc.catalogsApi()
+ .createCatalog(
+ new CreateCatalogRequest(testMechanismCatalog(name)),
+ svc.realmContext(),
+ svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void aStoredCatalogWhoseMechanismIsGoneServesMetadataAndRefusesToVend(boolean skipSubscoping) {
+ TestServices svc = services(config(skipSubscoping));
+ createTestMechanismCatalog(svc, "mechcat");
+ createNamespace(svc, "mechcat", "ns");
+ createTable(svc, "mechcat", "ns", "t");
+ createCatalog(svc, "emptycat");
+ createNamespace(svc, "emptycat", "ns");
+ createTable(svc, "emptycat", "ns", "t");
+ svc.installedMechanisms().remove(TEST_MECHANISM);
+
+ // Metadata routes keep serving: the mechanism is checked where it is stored and where it
+ // vends, never at catalog initialization.
+ try (Response r =
+ svc.restApi()
+ .listNamespaces(
+ "mechcat", null, null, null, svc.realmContext(), svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ try (Response r =
+ svc.restApi()
+ .loadNamespaceMetadata("mechcat", "ns", svc.realmContext(), svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ try (Response r =
+ svc.restApi()
+ .createNamespace(
+ "mechcat",
+ CreateNamespaceRequest.builder().withNamespace(Namespace.of("ns2")).build(),
+ null,
+ svc.realmContext(),
+ svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+
+ if (skipSubscoping) {
+ // The skip-subscoping early return in StorageAccessConfigProvider never reaches the
+ // registry, so vending succeeds even though the mechanism is gone.
+ assertLoadTableSucceeds(svc, "mechcat", "ns", "t", "vended-credentials");
+ } else {
+ assertLoadTableRefused(
+ svc,
+ "mechcat",
+ "ns",
+ "t",
+ "vended-credentials",
+ IllegalArgumentException.class,
+ NOT_AVAILABLE);
+ // buildLoadTableResponseWithDelegationCredentials calls StorageAccessConfigProvider on
+ // every load, delegation requested or not, so a plain load reaches the same registry gate.
+ assertLoadTableRefused(
+ svc, "mechcat", "ns", "t", null, IllegalArgumentException.class, NOT_AVAILABLE);
+ }
+
+ // Management reads are unaffected, and the catalog with an empty mechanism in the same realm
+ // still serves.
+ try (Response r =
+ svc.catalogsApi().getCatalog("mechcat", svc.realmContext(), svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ try (Response r =
+ svc.restApi()
+ .listNamespaces(
+ "emptycat", null, null, null, svc.realmContext(), svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ assertLoadTableSucceeds(svc, "emptycat", "ns", "t");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void theRealmKillSwitchRefusesTheCatalogWithNotEnabled(boolean skipSubscoping) {
+ Map<String, Object> config = config(skipSubscoping);
+ TestServices svc = services(config);
+ createTestMechanismCatalog(svc, "mechkill");
+ createNamespace(svc, "mechkill", "ns");
+ createTable(svc, "mechkill", "ns", "t");
+ createCatalog(svc, "emptykill");
+ createNamespace(svc, "emptykill", "ns");
+ createTable(svc, "emptykill", "ns", "t");
+
+ // Engage the kill switch: the realm no longer lists TEST_MECHANISM. The mechanism itself
+ // stays installed throughout.
+ config.put("SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS", List.of("STS"));
+
+ // Metadata routes keep serving: the mechanism is checked where it is stored and where it
+ // vends, never at catalog initialization.
+ try (Response r =
+ svc.restApi()
+ .listNamespaces(
+ "mechkill", null, null, null, svc.realmContext(), svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ try (Response r =
+ svc.restApi()
+ .loadNamespaceMetadata("mechkill", "ns", svc.realmContext(), svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ try (Response r =
+ svc.restApi()
+ .createNamespace(
+ "mechkill",
+ CreateNamespaceRequest.builder().withNamespace(Namespace.of("ns2")).build(),
+ null,
+ svc.realmContext(),
+ svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+
+ if (skipSubscoping) {
+ assertLoadTableSucceeds(svc, "mechkill", "ns", "t", "vended-credentials");
+ } else {
+ assertLoadTableRefused(
+ svc, "mechkill", "ns", "t", "vended-credentials", ValidationException.class, NOT_ENABLED);
+ assertLoadTableRefused(
+ svc, "mechkill", "ns", "t", null, ValidationException.class, NOT_ENABLED);
+ }
+
+ // Management reads still work; an update carrying a storage config is refused at site 1.
+ Catalog fetched;
+ try (Response r =
+ svc.catalogsApi().getCatalog("mechkill", svc.realmContext(), svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ fetched = (Catalog) r.getEntity();
+ }
+ UpdateCatalogRequest touch =
+ new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of("default-base-location", "s3://bucket/base/mechkill"),
+ fetched.getStorageConfigInfo());
+ assertThatThrownBy(
+ () ->
+ svc.catalogsApi()
+ .updateCatalog("mechkill", touch, svc.realmContext(), svc.securityContext()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(NOT_ENABLED);
+ // The catalog with an empty mechanism in the same realm is unaffected.
+ try (Response r =
+ svc.restApi()
+ .listNamespaces(
+ "emptykill", null, null, null, svc.realmContext(), svc.securityContext())) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ assertLoadTableSucceeds(svc, "emptykill", "ns", "t");
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/StorageConfigInfoJsonTest.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/StorageConfigInfoJsonTest.java
new file mode 100644
index 0000000000..ea6e02a0cd
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/StorageConfigInfoJsonTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.admin;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.exceptions.ValidationException;
+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.CreateCatalogRequest;
+import org.apache.polaris.core.admin.model.PolarisCatalog;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.service.TestServices;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The management API's JSON handling of {@code credentialVendingMechanism}: absent, null and
+ * unknown.
+ */
+class StorageConfigInfoJsonTest {
+
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ @Test
+ void absentAndNullMechanismDeserializeAsNull() throws Exception {
+ StorageConfigInfo absent =
+ mapper.readValue(
+ "{\"storageType\":\"S3\",\"allowedLocations\":[\"s3://b/p/\"]}",
+ StorageConfigInfo.class);
+ assertThat(((AwsStorageConfigInfo) absent).getCredentialVendingMechanism()).isNull();
+ StorageConfigInfo explicitNull =
+ mapper.readValue(
+ "{\"storageType\":\"S3\",\"allowedLocations\":[\"s3://b/p/\"],"
+ + "\"credentialVendingMechanism\":null}",
+ StorageConfigInfo.class);
+ assertThat(((AwsStorageConfigInfo) explicitNull).getCredentialVendingMechanism()).isNull();
+ }
+
+ @Test
+ void knownMechanismDeserializes() throws Exception {
+ StorageConfigInfo sts =
+ mapper.readValue(
+ "{\"storageType\":\"S3\",\"allowedLocations\":[\"s3://b/p/\"],"
+ + "\"credentialVendingMechanism\":\"STS\"}",
+ StorageConfigInfo.class);
+ assertThat(((AwsStorageConfigInfo) sts).getCredentialVendingMechanism()).isEqualTo("STS");
+ }
+
+ /**
+ * An unknown mechanism name is not a Jackson type error: the string field deserializes as is, and
+ * the realm allowlist refuses it at catalog create.
+ */
+ @Test
+ void unknownMechanismDeserializesAndIsRefusedByTheAllowlist() throws Exception {
+ String body =
+ "{\"storageType\":\"S3\",\"allowedLocations\":[\"s3://b/p/\"],"
+ + "\"credentialVendingMechanism\":\"BOGUS\"}";
+ StorageConfigInfo deserialized = mapper.readValue(body, StorageConfigInfo.class);
+ assertThat(((AwsStorageConfigInfo) deserialized).getCredentialVendingMechanism())
+ .isEqualTo("BOGUS");
+
+ TestServices svc =
+ TestServices.builder()
+ .config(Map.of("SUPPORTED_CATALOG_STORAGE_TYPES", List.of("S3")))
+ .build();
+ Catalog catalog =
+ PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName("bogus-mechanism")
+ .setProperties(new CatalogProperties("s3://bucket/base/"))
+ .setStorageConfigInfo(
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setCredentialVendingMechanism("BOGUS")
+ .setAllowedLocations(List.of("s3://bucket/base/"))
+ .build())
+ .build();
+ assertThatThrownBy(
+ () ->
+ svc.catalogsApi()
+ .createCatalog(
+ new CreateCatalogRequest(catalog),
+ svc.realmContext(),
+ svc.securityContext()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage("S3 credential vending mechanism BOGUS is not enabled in this realm");
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/policy/PolicyCatalogHandlerAuthzTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/policy/PolicyCatalogHandlerAuthzTest.java
index 44bcbe8f65..a5a5717449 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/policy/PolicyCatalogHandlerAuthzTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/policy/PolicyCatalogHandlerAuthzTest.java
@@ -40,6 +40,13 @@ import org.apache.polaris.service.types.UpdatePolicyRequest;
import org.junit.jupiter.api.DynamicNode;
import org.junit.jupiter.api.TestFactory;
+/**
+ * Privilege-matrix coverage for {@link PolicyCatalogHandler}. {@code initializeCatalog()} carries
+ * no S3 credential vending mechanism check: that mechanism is checked only where a catalog is
+ * stored (create and update) and where it vends a credential, covered end-to-end, through the real
+ * REST API and a stored catalog whose mechanism is removed from a live registry, by {@link
+ * org.apache.polaris.service.storage.S3CredentialVendingMechanismCdiTest#anUninstalledMechanismIsRefusedAtCreateAndUpdateAndAStoredOneWhenItVends}.
+ */
@QuarkusTest
@TestProfile(Profiles.PolarisAuthzBaseProfile.class)
public class PolicyCatalogHandlerAuthzTest extends PolarisAuthzTestBase {
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/validation/IcebergPropertiesValidationTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/validation/IcebergPropertiesValidationTest.java
new file mode 100644
index 0000000000..205a407577
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/validation/IcebergPropertiesValidationTest.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.catalog.validation;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class IcebergPropertiesValidationTest {
+
+ @Mock private RealmConfig realmConfig;
+
+ private void allow(String... mechanisms) {
+ when(realmConfig.getConfig(eq(FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS)))
+ .thenReturn(List.of(mechanisms));
+ }
+
+ @Test
+ void defaultAllowlistIsStsOnly() {
+ when(realmConfig.getConfig(eq(FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS)))
+ .thenReturn(FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS.defaultValue());
+ assertThatCode(
+ () ->
+ IcebergPropertiesValidation.validateS3CredentialVendingMechanismAllowed(
+ realmConfig, "STS"))
+ .doesNotThrowAnyException();
+ assertThatThrownBy(
+ () ->
+ IcebergPropertiesValidation.validateS3CredentialVendingMechanismAllowed(
+ realmConfig, "SECOND_MECHANISM"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(
+ "S3 credential vending mechanism SECOND_MECHANISM is not enabled in this realm");
+ }
+
+ @Test
+ void theListHasNoImplicitMember() {
+ allow("SECOND_MECHANISM");
+ assertThatThrownBy(
+ () ->
+ IcebergPropertiesValidation.validateS3CredentialVendingMechanismAllowed(
+ realmConfig, "STS"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage("S3 credential vending mechanism STS is not enabled in this realm");
+ }
+
+ @Test
+ void anEmptyMechanismSkipsTheAllowlistAndResolvesToSts() {
+ allow("SECOND_MECHANISM"); // would refuse STS if consulted
+ AwsStorageConfigurationInfo empty =
+ AwsStorageConfigurationInfo.builder().addAllowedLocation("s3://bucket/prefix/").build();
+ assertThatCode(
+ () ->
+ IcebergPropertiesValidation.validateS3CredentialVendingMechanismAllowed(
+ realmConfig, (String) null))
+ .doesNotThrowAnyException();
+ assertThat(empty.resolvedCredentialVendingMechanism())
+ .isEqualTo(S3CredentialVendingMechanism.STS);
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/config/ProductionReadinessChecksTest.java b/runtime/service/src/test/java/org/apache/polaris/service/config/ProductionReadinessChecksTest.java
index fe0ccde3ea..3e0923068a 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/config/ProductionReadinessChecksTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/config/ProductionReadinessChecksTest.java
@@ -19,19 +19,25 @@
package org.apache.polaris.service.config;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
+import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
+import org.apache.polaris.core.config.FeatureConfiguration;
import org.apache.polaris.core.config.ProductionReadinessCheck;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
import org.apache.polaris.service.auth.AuthenticationConfiguration;
import org.apache.polaris.service.auth.AuthenticationRealmConfiguration;
import org.apache.polaris.service.auth.AuthenticationType;
import org.apache.polaris.service.auth.CredentialMode;
import org.apache.polaris.service.auth.external.OidcConfiguration;
import org.apache.polaris.service.auth.external.tenant.OidcTenantConfiguration;
+import org.apache.polaris.service.storage.S3CredentialVendingMechanisms;
import org.eclipse.microprofile.config.Config;
import org.eclipse.microprofile.config.ConfigValue;
import org.junit.jupiter.api.BeforeEach;
@@ -380,4 +386,116 @@ class ProductionReadinessChecksTest {
when(configValue.getValue()).thenReturn(value);
return config;
}
+
+ private static final String MECHANISMS_KEY =
+ FeatureConfiguration.SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS.key();
+
+ private static S3CredentialVendingMechanisms installed(String... ids) {
+ Map<String, S3CredentialVendingMechanism> mechanisms = new HashMap<>();
+ for (String id : ids) {
+ mechanisms.put(id, mock(S3CredentialVendingMechanism.class));
+ }
+ return new S3CredentialVendingMechanisms(mechanisms);
+ }
+
+ private static FeaturesConfiguration featuresConfig(
+ Map<String, String> defaults, Map<String, RealmOverridable.RealmOverrides> realmOverrides) {
+ // CALLS_REAL_METHODS keeps the interface's default parseDefaults/parseRealmOverrides working.
+ FeaturesConfiguration config =
+ mock(FeaturesConfiguration.class, withSettings().defaultAnswer(CALLS_REAL_METHODS));
+ when(config.defaults()).thenReturn(defaults);
+ when(config.realmOverrides()).thenReturn(realmOverrides);
+ return config;
+ }
+
+ private static RealmOverridable.RealmOverrides overrides(Map<String, String> values) {
+ RealmOverridable.RealmOverrides realmOverrides = mock(RealmOverridable.RealmOverrides.class);
+ when(realmOverrides.overrides()).thenReturn(values);
+ return realmOverrides;
+ }
+
+ @Test
+ void everyAllowlistedMechanismInstalledIsReady() {
+ ProductionReadinessCheck result =
+ checks.checkS3CredentialVendingMechanisms(
+ featuresConfig(
+ Map.of(MECHANISMS_KEY, "[\"STS\"]"),
+ Map.of("r1", overrides(Map.of(MECHANISMS_KEY, "[\"STS\"]")))),
+ installed("STS"));
+ assertThat(result.ready()).isTrue();
+ }
+
+ @Test
+ void anAllowlistedUninstalledMechanismInDefaultsIsNonSevereAndNamesWhatIsAvailable() {
+ ProductionReadinessCheck result =
+ checks.checkS3CredentialVendingMechanisms(
+ featuresConfig(Map.of(MECHANISMS_KEY, "[\"STS\",\"BOGUS\"]"), Map.of()),
+ installed("STS"));
+ assertThat(result.getErrors())
+ .singleElement()
+ .satisfies(
+ error -> {
+ assertThat(error.severe()).isFalse();
+ assertThat(error.offendingProperty())
+ .isEqualTo("polaris.features.\"" + MECHANISMS_KEY + "\"");
+ assertThat(error.message()).contains("BOGUS").contains("STS");
+ });
+ }
+
+ @Test
+ void anUnconfiguredAllowlistChecksTheCodeDefault() {
+ ProductionReadinessCheck result =
+ checks.checkS3CredentialVendingMechanisms(
+ featuresConfig(Map.of(), Map.of()), installed("SECOND"));
+ assertThat(result.getErrors())
+ .filteredOn(error -> !error.severe())
+ .singleElement()
+ .satisfies(
+ error -> {
+ assertThat(error.offendingProperty())
+ .isEqualTo("polaris.features.\"" + MECHANISMS_KEY + "\"");
+ assertThat(error.message()).contains("STS");
+ });
+ }
+
+ @Test
+ void anUnconfiguredAllowlistIsReadyWhenStsIsInstalled() {
+ ProductionReadinessCheck result =
+ checks.checkS3CredentialVendingMechanisms(
+ featuresConfig(Map.of(), Map.of()), installed("STS"));
+ assertThat(result.ready()).isTrue();
+ }
+
+ @Test
+ void anAllowlistedUninstalledMechanismInARealmOverrideNamesTheRealm() {
+ ProductionReadinessCheck result =
+ checks.checkS3CredentialVendingMechanisms(
+ featuresConfig(Map.of(), Map.of("r1", overrides(Map.of(MECHANISMS_KEY, "[\"NOPE\"]")))),
+ installed("STS"));
+ assertThat(result.getErrors())
+ .singleElement()
+ .satisfies(
+ error -> {
+ assertThat(error.severe()).isFalse();
+ assertThat(error.offendingProperty()).contains("r1").contains(MECHANISMS_KEY);
+ assertThat(error.message()).contains("NOPE").contains("STS");
+ });
+ }
+
+ @Test
+ void aMissingStsMechanismIsASevereError() {
+ ProductionReadinessCheck result =
+ checks.checkS3CredentialVendingMechanisms(
+ featuresConfig(Map.of(MECHANISMS_KEY, "[\"SECOND\"]"), Map.of()), installed("SECOND"));
+ assertThat(result.ready()).isFalse();
+ assertThat(result.getErrors())
+ .singleElement()
+ .satisfies(
+ error -> {
+ assertThat(error.severe()).isTrue();
+ assertThat(error.message()).contains("STS");
+ assertThat(error.offendingProperty())
+ .isEqualTo("S3CredentialVendingMechanism @Identifier(\"STS\")");
+ });
+ }
}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/entity/CatalogEntityTest.java b/runtime/service/src/test/java/org/apache/polaris/service/entity/CatalogEntityTest.java
index 2998d0c497..44c10087dd 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/entity/CatalogEntityTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/entity/CatalogEntityTest.java
@@ -48,6 +48,7 @@ import org.apache.polaris.core.entity.CatalogEntity;
import org.apache.polaris.core.identity.credential.AwsIamServiceIdentityCredential;
import org.apache.polaris.core.identity.dpo.AwsIamServiceIdentityInfoDpo;
import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
+import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -536,6 +537,31 @@ public class CatalogEntityTest {
assertThat(MAPPER.writeValueAsString(catalog.getStorageConfigInfo())).isEqualTo(configStr);
}
+ @Test
+ public void testBlankMechanismInTheModelIsStoredAsEmpty() {
+ AwsStorageConfigInfo blank =
+ AwsStorageConfigInfo.builder()
+ .setStorageType(StorageConfigInfo.StorageTypeEnum.S3)
+ .setAllowedLocations(List.of("s3://bucket/path/"))
+ .setRoleArn("arn:aws:iam::012345678901:role/test-role")
+ .setCredentialVendingMechanism(" ")
+ .build();
+ CatalogEntity entity =
+ new CatalogEntity.Builder()
+ .setName("blank-mechanism")
+ .setDefaultBaseLocation("s3://bucket/path/")
+ .setStorageConfigurationInfo(realmConfig, blank)
+ .build();
+ AwsStorageConfigurationInfo stored =
+ (AwsStorageConfigurationInfo) entity.getStorageConfigurationInfo();
+ assertThat(stored.getCredentialVendingMechanism()).isNull();
+ assertThat(
+ ((AwsStorageConfigInfo)
+ entity.asCatalog(serviceIdentityProvider).getStorageConfigInfo())
+ .getCredentialVendingMechanism())
+ .isNull();
+ }
+
@Test
public void testServiceIdentityInjection() {
String baseLocation = "s3://test-bucket/path";
@@ -614,6 +640,7 @@ public class CatalogEntityTest {
Arguments.of(b.setStsEndpoint("http://sts.example.com:1234").build()),
Arguments.of(b.setPathStyleAccess(true).build()),
Arguments.of(b.setStorageName("my-storage").build()),
+ Arguments.of(b.setCredentialVendingMechanism("STS").build()),
Arguments.of(a.build()),
Arguments.of(a.setHierarchical(true).build()));
}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/RecordingStsCredentialVendingMechanism.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/RecordingStsCredentialVendingMechanism.java
new file mode 100644
index 0000000000..1d1a088b33
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/RecordingStsCredentialVendingMechanism.java
@@ -0,0 +1,70 @@
+/*
+ * 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;
+
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Alternative;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.apache.polaris.core.storage.PolarisStorageIntegration;
+import org.apache.polaris.core.storage.StorageAccessConfig;
+import org.apache.polaris.core.storage.StorageAccessProperty;
+import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
+
+/**
+ * A recording replacement for the STS mechanism, enabled only under {@link
+ * RecordingStsMechanismProfile}: proves that an enabled alternative carrying the STS identifier
+ * serves both an empty credentialVendingMechanism and an explicit STS.
+ */
+@ApplicationScoped
+@Alternative
+@Identifier(S3CredentialVendingMechanism.STS)
+public class RecordingStsCredentialVendingMechanism implements S3CredentialVendingMechanism {
+
+ public static final String FAKE_KEY_FOR_TEST = "STS_OVERRIDE_FAKE_KEY";
+ public static final String FAKE_SECRET_FOR_TEST = "STS_OVERRIDE_FAKE_SECRET";
+ public static final String FAKE_TOKEN_FOR_TEST = "STS_OVERRIDE_FAKE_TOKEN";
+
+ /** One call: the storage config {@link #integrationFor} was given. */
+ public record Call(AwsStorageConfigurationInfo storageConfig) {}
+
+ private final List<Call> calls = new CopyOnWriteArrayList<>();
+
+ public List<Call> calls() {
+ return List.copyOf(calls);
+ }
+
+ public void clear() {
+ calls.clear();
+ }
+
+ @Override
+ public PolarisStorageIntegration integrationFor(AwsStorageConfigurationInfo storageConfig) {
+ calls.add(new Call(storageConfig));
+ return (grants, refreshEndpoint, context) ->
+ StorageAccessConfig.builder()
+ .putCredential(StorageAccessProperty.AWS_KEY_ID.getPropertyName(), FAKE_KEY_FOR_TEST)
+ .putCredential(
+ StorageAccessProperty.AWS_SECRET_KEY.getPropertyName(), FAKE_SECRET_FOR_TEST)
+ .putCredential(StorageAccessProperty.AWS_TOKEN.getPropertyName(), FAKE_TOKEN_FOR_TEST)
+ .build();
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/RecordingStsMechanismProfile.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/RecordingStsMechanismProfile.java
new file mode 100644
index 0000000000..57990bf998
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/RecordingStsMechanismProfile.java
@@ -0,0 +1,55 @@
+/*
+ * 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;
+
+import io.quarkus.test.junit.QuarkusTestProfile;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Installs {@link RecordingStsCredentialVendingMechanism} as a CDI alternative for the STS
+ * identifier. Only {@code STS} is allowlisted explicitly: an empty {@code
+ * credentialVendingMechanism}, which resolves to STS, needs no allowlist entry of its own.
+ */
+public class RecordingStsMechanismProfile implements QuarkusTestProfile {
+
+ @Override
+ public Set<Class<?>> getEnabledAlternatives() {
+ return Set.of(RecordingStsCredentialVendingMechanism.class);
+ }
+
+ @Override
+ public Map<String, String> getConfigOverrides() {
+ return Map.of(
+ "polaris.realm-context.realms",
+ ThirdMechanismProfile.ALLOWLISTED_REALM,
+ "polaris.file-io.type",
+ "test-in-memory",
+ "polaris.features.\"SUPPORTED_CATALOG_STORAGE_TYPES\"",
+ "[\"S3\"]",
+ "polaris.features.\"SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS\"",
+ "[\"STS\"]",
+ "polaris.event-listener.type",
+ "test",
+ "polaris.authentication.token-broker.type",
+ "symmetric-key",
+ "polaris.authentication.token-broker.symmetric-key.secret",
+ "secret");
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/RecordingThirdCredentialVendingMechanism.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/RecordingThirdCredentialVendingMechanism.java
new file mode 100644
index 0000000000..0f6aa81238
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/RecordingThirdCredentialVendingMechanism.java
@@ -0,0 +1,96 @@
+/*
+ * 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;
+
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Alternative;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.apache.polaris.core.storage.PolarisStorageIntegration;
+import org.apache.polaris.core.storage.StorageAccessConfig;
+import org.apache.polaris.core.storage.StorageAccessProperty;
+import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * A recording third {@link S3CredentialVendingMechanism}, installed only under {@link
+ * ThirdMechanismProfile}, that records every call it receives and vends a fixed, fake AWS
+ * credential triple. Proves that the registry and the gates work for a mechanism the server itself
+ * does not ship, not only for {@code STS}.
+ */
+@ApplicationScoped
+@Alternative
+@Identifier(RecordingThirdCredentialVendingMechanism.ID)
+public class RecordingThirdCredentialVendingMechanism implements S3CredentialVendingMechanism {
+
+ public static final String ID = "TEST_MECHANISM";
+ public static final String FAKE_KEY_FOR_TEST = "TEST_MECHANISM_FAKE_KEY";
+ public static final String FAKE_SECRET_FOR_TEST = "TEST_MECHANISM_FAKE_SECRET";
+ public static final String FAKE_TOKEN_FOR_TEST = "TEST_MECHANISM_FAKE_TOKEN";
+
+ /** One call: the storage config {@link #integrationFor} was given. */
+ public record Call(AwsStorageConfigurationInfo storageConfig) {}
+
+ private final List<Call> calls = new CopyOnWriteArrayList<>();
+
+ public List<Call> calls() {
+ return List.copyOf(calls);
+ }
+
+ /** One validation: the stored config, null at create, and the config being written. */
+ public record Validation(
+ @Nullable AwsStorageConfigurationInfo current, AwsStorageConfigurationInfo updated) {}
+
+ private final List<Validation> validations = new CopyOnWriteArrayList<>();
+
+ public List<Validation> validations() {
+ return List.copyOf(validations);
+ }
+
+ public void clear() {
+ calls.clear();
+ validations.clear();
+ }
+
+ @Override
+ public PolarisStorageIntegration integrationFor(AwsStorageConfigurationInfo storageConfig) {
+ calls.add(new Call(storageConfig));
+ return (grants, refreshEndpoint, context) ->
+ StorageAccessConfig.builder()
+ .putCredential(StorageAccessProperty.AWS_KEY_ID.getPropertyName(), FAKE_KEY_FOR_TEST)
+ .putCredential(
+ StorageAccessProperty.AWS_SECRET_KEY.getPropertyName(), FAKE_SECRET_FOR_TEST)
+ .putCredential(StorageAccessProperty.AWS_TOKEN.getPropertyName(), FAKE_TOKEN_FOR_TEST)
+ .build();
+ }
+
+ @Override
+ public void validate(
+ @Nullable AwsStorageConfigurationInfo current, AwsStorageConfigurationInfo updated) {
+ validations.add(new Validation(current, updated));
+ for (String location : updated.getAllowedLocations()) {
+ if (location.contains("/refused/")) {
+ throw new IllegalArgumentException(
+ "TEST_MECHANISM refuses the allowed location " + location);
+ }
+ }
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismCdiTest.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismCdiTest.java
new file mode 100644
index 0000000000..1092e8434c
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismCdiTest.java
@@ -0,0 +1,572 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.quarkus.test.junit.QuarkusMock;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import io.quarkus.test.junit.TestProfile;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.client.Entity;
+import jakarta.ws.rs.core.Response;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.responses.ErrorResponse;
+import org.apache.iceberg.rest.responses.ErrorResponseParser;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.CatalogGrant;
+import org.apache.polaris.core.admin.model.CatalogPrivilege;
+import org.apache.polaris.core.admin.model.CatalogProperties;
+import org.apache.polaris.core.admin.model.CreateCatalogRequest;
+import org.apache.polaris.core.admin.model.GrantResource;
+import org.apache.polaris.core.admin.model.PolarisCatalog;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.admin.model.UpdateCatalogRequest;
+import org.apache.polaris.core.entity.PolarisEntityConstants;
+import org.apache.polaris.core.storage.StorageAccessProperty;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
+import org.apache.polaris.service.TestServices;
+import org.apache.polaris.service.admin.PolarisAuthzTestBase;
+import org.apache.polaris.service.it.env.CatalogApi;
+import org.apache.polaris.service.it.env.ClientCredentials;
+import org.apache.polaris.service.it.env.GenericTableApi;
+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.service.it.env.PolicyApi;
+import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension;
+import org.jspecify.annotations.Nullable;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * CDI-level coverage for {@link S3CredentialVendingMechanisms}: discovery through the real
+ * container, and the standalone contract this module alone must satisfy: an allowlisted mechanism
+ * with no bean installed is refused at create, at update, and whenever a stored catalog that
+ * selected it vends a credential, but keeps serving the catalog's metadata routes, never confused
+ * with STS, and never aborts startup, while the STS bean is proven to still vend through the real
+ * registry dispatch, for an empty field and for an explicit STS. No cloud calls: every catalog's
+ * table content goes through {@link TestInMemoryFileIOFactory} (selected by {@code
+ * polaris.file-io.type=test-in-memory}).
+ *
+ * <p>The container cannot uninstall a bean, so the first test method installs, for its own duration
+ * only, a registry over a live, mutable map through {@link QuarkusMock#installMockForType}: a
+ * catalog is created selecting {@link #TEST_MECHANISM}, a fake mechanism this test adds to that
+ * map, then the map loses the mechanism, which is what a server rebuilt without the mechanism looks
+ * like to a catalog that already stored it. Quarkus restores the real registry after the test
+ * method. The namespace and table are created while the mechanism is still installed, before the
+ * map loses it, so only the later credential-vending call is exercised against the gone mechanism.
+ *
+ * <p>The "not available in this server" assertions against a mechanism identifier the server never
+ * ships at all, {@code UNINSTALLED_MECHANISM}, cover create-time and update-time refusal; the ones
+ * against {@link #TEST_MECHANISM} cover a mechanism that was installed when a catalog selected it
+ * and is later removed from the live registry.
+ *
+ * <p>This class installs only the {@code STS} mechanism (the server's real, shipped bean) plus, for
+ * the first test method's own live registry, a fake it adds itself, and allowlists {@code
+ * UNINSTALLED_MECHANISM} without ever installing it. {@link
+ * S3CredentialVendingMechanismThirdMechanismCdiTest} swaps in a test-only mechanism as a real CDI
+ * bean, under its own profile, to prove the registry and the gates work for a mechanism the server
+ * itself does not ship. Quarkus does not allow {@code @TestProfile} on a {@code @Nested} class, so
+ * that scenario cannot share this file: it needs its own application instance, since {@code
+ * getEnabledAlternatives()} is a profile-wide, one-instance setting, and this class asserts {@code
+ * availableIds()} is exactly {@code {STS}}.
+ */
+@QuarkusTest
+@TestProfile(S3CredentialVendingMechanismCdiTest.Profile.class)
+@ExtendWith(PolarisIntegrationTestExtension.class)
+class S3CredentialVendingMechanismCdiTest {
+
+ private static final String UNINSTALLED_MECHANISM = "UNINSTALLED_MECHANISM";
+ private static final String NOT_AVAILABLE_UNINSTALLED =
+ "S3 credential vending mechanism UNINSTALLED_MECHANISM is not available in this server";
+ private static final String TEST_MECHANISM = "TEST_MECHANISM";
+ private static final String NOT_AVAILABLE_TEST =
+ "S3 credential vending mechanism TEST_MECHANISM is not available in this server";
+
+ /** A second realm, used only to run the STS-vending check with the skip flag turned off. */
+ private static final String NO_SKIP_REALM = "POLARIS2";
+
+ public static class Profile implements QuarkusTestProfile {
+ @Override
+ public Map<String, String> getConfigOverrides() {
+ return Map.ofEntries(
+ Map.entry("polaris.realm-context.realms", "POLARIS," + NO_SKIP_REALM),
+ Map.entry("polaris.file-io.type", "test-in-memory"),
+ Map.entry("polaris.features.\"SUPPORTED_CATALOG_STORAGE_TYPES\"", "[\"S3\"]"),
+ Map.entry("polaris.features.\"ENABLE_GENERIC_TABLES\"", "true"),
+ Map.entry("polaris.features.\"ENABLE_POLICY_STORE\"", "true"),
+ Map.entry("polaris.features.\"SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION\"", "true"),
+ Map.entry(
+ "polaris.features.realm-overrides.\""
+ + NO_SKIP_REALM
+ + "\".\"SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION\"",
+ "false"),
+ Map.entry(
+ "polaris.features.\"SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS\"",
+ "[\"STS\",\"" + TEST_MECHANISM + "\",\"" + UNINSTALLED_MECHANISM + "\"]"),
+ Map.entry("polaris.event-listener.type", "test"),
+ Map.entry("polaris.authentication.token-broker.type", "symmetric-key"),
+ Map.entry("polaris.authentication.token-broker.symmetric-key.secret", "secret"));
+ }
+ }
+
+ @Inject S3CredentialVendingMechanisms mechanisms;
+
+ @Inject
+ @Identifier(S3CredentialVendingMechanism.STS)
+ S3CredentialVendingMechanism stsBean;
+
+ /**
+ * With the default readiness settings (no {@code polaris.readiness.ignore-severe-issues}
+ * override) and two mechanisms allowlisted but not installed, the application starts, only {@code
+ * STS} is available on the real registry, and: creating or updating a catalog to select {@code
+ * UNINSTALLED_MECHANISM} is refused with "not available in this server"; a catalog created while
+ * {@link #TEST_MECHANISM} is installed in a live registry this test installs over the real one
+ * keeps serving its metadata routes once that registry loses the mechanism, the Iceberg,
+ * generic-table and policy routes alike, and refuses only a plain load, which reaches the
+ * registry through {@link org.apache.polaris.service.catalog.io.StorageAccessConfigProvider}. A
+ * catalog with an empty mechanism in the same realm is untouched throughout.
+ */
+ @Test
+ void anUninstalledMechanismIsRefusedAtCreateAndUpdateAndAStoredOneWhenItVends(
+ PolarisApiEndpoints endpoints, ClientCredentials credentials) throws Exception {
+ // The application started at all, with two mechanisms allowlisted and neither installed, and
+ // default readiness settings, is itself part of what this proves; availableIds() shows only
+ // STS installed.
+ assertThat(mechanisms.availableIds()).containsExactly("STS");
+
+ Map<String, S3CredentialVendingMechanism> live =
+ new ConcurrentHashMap<>(
+ Map.of(
+ S3CredentialVendingMechanism.STS,
+ stsBean,
+ TEST_MECHANISM,
+ TestServices.fakeMechanism()));
+ QuarkusMock.installMockForType(
+ new S3CredentialVendingMechanisms(live), S3CredentialVendingMechanisms.class);
+
+ try (PolarisClient client = PolarisClient.polarisClient(endpoints)) {
+ String adminToken = client.obtainToken(credentials);
+ ManagementApi managementApi = client.managementApi(adminToken);
+ CatalogApi catalogApi = client.catalogApi(adminToken);
+ GenericTableApi genericTableApi = client.genericTableApi(adminToken);
+ PolicyApi policyApi = client.policyApi(adminToken);
+
+ String emptyMechanismCatalog = "cdi-empty-cat";
+
+ // The catalog with an empty mechanism: content created normally, left untouched for the
+ // rest of the test.
+ createEmptyMechanismCatalog(managementApi, emptyMechanismCatalog);
+ try (Response r =
+ managementApi
+ .request("v1/catalogs/{name}", Map.of("name", emptyMechanismCatalog))
+ .get()) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ assertThat(r.readEntity(String.class)).doesNotContain("credentialVendingMechanism");
+ }
+ catalogApi.createNamespace(emptyMechanismCatalog, "ns");
+ createTable(catalogApi, emptyMechanismCatalog, "ns", "t");
+
+ // Creating a catalog that selects a mechanism the server never ships is refused before any
+ // content exists.
+ try (Response refused =
+ managementApi
+ .request("v1/catalogs")
+ .post(
+ Entity.json(
+ new CreateCatalogRequest(
+ uninstalledMechanismCatalog("cdi-create-uninstalled-cat"))))) {
+ assertRefused(refused, NOT_AVAILABLE_UNINSTALLED);
+ }
+
+ // Switching the catalog with an empty mechanism to the same mechanism at update is refused
+ // the same way.
+ try (Response refused =
+ managementApi
+ .request("v1/catalogs/{name}", Map.of("name", emptyMechanismCatalog))
+ .put(
+ Entity.json(
+ updateToUninstalledMechanismRequest(managementApi, emptyMechanismCatalog)))) {
+ assertRefused(refused, NOT_AVAILABLE_UNINSTALLED);
+ }
+
+ // A catalog created in NO_SKIP_REALM (skip-subscoping off, so a plain load reaches the
+ // registry) while TEST_MECHANISM is installed in the live registry this test installed
+ // above, populated with a namespace and table, then the registry loses the mechanism.
+ String testMechCatalog = "cdi-test-mech-cat";
+ Map<String, String> noSkipRealmHeaders =
+ Map.of("Authorization", "Bearer " + adminToken, "Polaris-Realm", NO_SKIP_REALM);
+ createTestMechanismCatalog(managementApi, noSkipRealmHeaders, testMechCatalog);
+ try (Response r =
+ catalogApi
+ .request(
+ "v1/{cat}/namespaces",
+ Map.of("cat", testMechCatalog),
+ Map.of(),
+ noSkipRealmHeaders)
+ .post(
+ Entity.json(
+ CreateNamespaceRequest.builder()
+ .withNamespace(Namespace.of("ns"))
+ .build()))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ try (Response r =
+ catalogApi
+ .request(
+ "v1/{cat}/namespaces/{ns}/tables",
+ Map.of("cat", testMechCatalog, "ns", "ns"),
+ Map.of(),
+ noSkipRealmHeaders)
+ .post(
+ Entity.json(
+ CreateTableRequest.builder()
+ .withName("t")
+ .withSchema(PolarisAuthzTestBase.SCHEMA)
+ .withLocation("s3://bucket/base/" + testMechCatalog + "/ns/t/")
+ .build()))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+
+ live.remove(TEST_MECHANISM);
+
+ // Metadata routes keep serving: the mechanism is checked where it is stored and where it
+ // vends, never at catalog initialization.
+ try (Response r =
+ catalogApi
+ .request(
+ "v1/{cat}/namespaces",
+ Map.of("cat", testMechCatalog),
+ Map.of(),
+ noSkipRealmHeaders)
+ .get()) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ try (Response r =
+ genericTableApi
+ .request(
+ "polaris/v1/{cat}/namespaces/{ns}/generic-tables",
+ Map.of("cat", testMechCatalog, "ns", "ns"),
+ Map.of(),
+ noSkipRealmHeaders)
+ .get()) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ try (Response r =
+ policyApi
+ .request(
+ "polaris/v1/{cat}/namespaces/{ns}/policies",
+ Map.of("cat", testMechCatalog, "ns", "ns"),
+ Map.of(),
+ noSkipRealmHeaders)
+ .get()) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ // A plain load still reaches the registry: buildLoadTableResponseWithDelegationCredentials
+ // calls StorageAccessConfigProvider unconditionally on every load, delegation requested or
+ // not, which resolves the catalog's mechanism through the provider.
+ assertRefused(
+ catalogApi
+ .request(
+ "v1/{cat}/namespaces/{ns}/tables/{table}",
+ Map.of("cat", testMechCatalog, "ns", "ns", "table", "t"),
+ Map.of(),
+ noSkipRealmHeaders)
+ .get(),
+ NOT_AVAILABLE_TEST);
+
+ // The catalog with an empty mechanism in the same realm is unaffected throughout.
+ try (Response ok =
+ catalogApi.request("v1/{cat}/namespaces", Map.of("cat", emptyMechanismCatalog)).get()) {
+ assertThat(ok.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ try (Response ok =
+ catalogApi
+ .request(
+ "v1/{cat}/namespaces/{ns}/tables/{table}",
+ Map.of("cat", emptyMechanismCatalog, "ns", "ns", "table", "t"))
+ .get()) {
+ assertThat(ok.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ }
+ }
+
+ /**
+ * Proves the second half of STS discovery that the other test method cannot: that the STS bean
+ * actually vends through the registry, not merely that the gate admits it. That test method's
+ * realm has {@code SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION} on, so {@code
+ * StorageAccessConfigProvider} returns before ever calling the registry; its 200s show only that
+ * the route works. This method runs in a realm with the skip flag off, on catalogs with {@code
+ * stsUnavailable: true}, so no real STS call is made: {@code
+ * buildLoadTableResponseWithDelegationCredentials} calls {@code StorageAccessConfigProvider}
+ * unconditionally on every load, delegation requested or not, which reaches the registry, which
+ * resolves the catalog's mechanism (an absent value and an explicit one both resolving to
+ * {@code @Identifier("STS")}) and runs {@code AwsCredentialsStorageIntegration}, which checks
+ * {@code stsUnavailable}, skips the AssumeRole call entirely, and returns only the catalog's
+ * non-credential storage properties (its region and endpoint). A load succeeding with those
+ * properties present and no access key anywhere in the response is only possible if the registry
+ * actually resolved the bean and ran its integration. The sequence runs once for each mechanism,
+ * on its own catalog. This method runs against the real registry: Quarkus restores it after the
+ * previous test method installed a mock over it for its own duration only.
+ */
+ @Test
+ void theStsMechanismVendsThroughTheRegistryForAnEmptyAndAnExplicitFieldWithoutARealStsCall(
+ PolarisApiEndpoints endpoints, ClientCredentials credentials) throws Exception {
+ try (PolarisClient client = PolarisClient.polarisClient(endpoints)) {
+ String adminToken = client.obtainToken(credentials);
+ ManagementApi managementApi = client.managementApi(adminToken);
+ CatalogApi catalogApi = client.catalogApi(adminToken);
+ Map<String, String> realmHeaders =
+ Map.of("Authorization", "Bearer " + adminToken, "Polaris-Realm", NO_SKIP_REALM);
+
+ String endpoint = "https://s3.example-compatible-store.test";
+ String region = "us-east-1";
+
+ Map<String, String> scenarios = new LinkedHashMap<>();
+ scenarios.put("cdi-empty-unavailable-cat", null);
+ scenarios.put("cdi-sts-unavailable-cat", "STS");
+
+ for (Map.Entry<String, String> scenario : scenarios.entrySet()) {
+ String catalog = scenario.getKey();
+ String mechanism = scenario.getValue();
+
+ createStsUnavailableCatalog(
+ managementApi, realmHeaders, catalog, mechanism, endpoint, region);
+
+ // The catalog's auto-granted catalog_admin role carries CATALOG_MANAGE_ACCESS and
+ // CATALOG_MANAGE_METADATA only. This grant broadens it to CATALOG_MANAGE_CONTENT, the same
+ // grant ManagementApi.makeAdmin gives a caller-supplied catalog role.
+ try (Response r =
+ managementApi
+ .request(
+ "v1/catalogs/{cat}/catalog-roles/{role}/grants",
+ Map.of(
+ "cat", catalog, "role", PolarisEntityConstants.getNameOfCatalogAdminRole()),
+ Map.of(),
+ realmHeaders)
+ .put(
+ Entity.json(
+ new CatalogGrant(
+ CatalogPrivilege.CATALOG_MANAGE_CONTENT,
+ GrantResource.TypeEnum.CATALOG)))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+
+ try (Response r =
+ catalogApi
+ .request("v1/{cat}/namespaces", Map.of("cat", catalog), Map.of(), realmHeaders)
+ .post(
+ Entity.json(
+ CreateNamespaceRequest.builder()
+ .withNamespace(Namespace.of("ns"))
+ .build()))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+
+ try (Response r =
+ catalogApi
+ .request(
+ "v1/{cat}/namespaces/{ns}/tables",
+ Map.of("cat", catalog, "ns", "ns"),
+ Map.of(),
+ realmHeaders)
+ .post(
+ Entity.json(
+ CreateTableRequest.builder()
+ .withName("t")
+ .withSchema(PolarisAuthzTestBase.SCHEMA)
+ .withLocation("s3://bucket/base/" + catalog + "/ns/t/")
+ .build()))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+
+ // No X-Iceberg-Access-Delegation header: buildLoadTableResponseWithDelegationCredentials
+ // calls StorageAccessConfigProvider unconditionally for every loadTable, delegation
+ // requested or not, so this still reaches the registry and the resolved bean's
+ // integration. Requesting delegation here does not work hermetically: a single
+ // "vended-credentials" mode with no credentials available throws ("Credential vending
+ // was requested... but no credentials are available"), and requesting both modes
+ // together resolves to remote-signing once STS is unavailable, which
+ // IcebergCatalogHandler rejects outright as not yet implemented. A plain load avoids
+ // both and still proves the same thing: the registry resolved the bean and ran its
+ // integration to produce these properties.
+ try (Response r =
+ catalogApi
+ .request(
+ "v1/{cat}/namespaces/{ns}/tables/{table}",
+ Map.of("cat", catalog, "ns", "ns", "table", "t"),
+ Map.of(),
+ realmHeaders)
+ .get()) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ LoadTableResponse loaded = r.readEntity(LoadTableResponse.class);
+ assertThat(loaded.credentials()).isEmpty();
+ assertThat(loaded.config())
+ .containsEntry(StorageAccessProperty.CLIENT_REGION.getPropertyName(), region)
+ .containsEntry(StorageAccessProperty.AWS_ENDPOINT.getPropertyName(), endpoint)
+ .doesNotContainKey(StorageAccessProperty.AWS_KEY_ID.getPropertyName());
+ }
+ }
+ }
+ }
+
+ private static void createStsUnavailableCatalog(
+ ManagementApi managementApi,
+ Map<String, String> realmHeaders,
+ String name,
+ @Nullable String mechanism,
+ String endpoint,
+ String region) {
+ AwsStorageConfigInfo.Builder storageConfig =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setStsUnavailable(true)
+ .setEndpoint(endpoint)
+ .setPathStyleAccess(true)
+ .setRegion(region)
+ .setAllowedLocations(List.of("s3://bucket/base/" + name + "/"));
+ if (mechanism != null) {
+ storageConfig.setCredentialVendingMechanism(mechanism);
+ }
+ try (Response r =
+ managementApi
+ .request("v1/catalogs", Map.of(), Map.of(), realmHeaders)
+ .post(
+ Entity.json(
+ new CreateCatalogRequest(
+ PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(name)
+ .setProperties(new CatalogProperties("s3://bucket/base/" + name))
+ .setStorageConfigInfo(storageConfig.build())
+ .build())))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ }
+
+ private static void createEmptyMechanismCatalog(ManagementApi managementApi, String name) {
+ managementApi.createCatalog(
+ PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(name)
+ .setProperties(new CatalogProperties("s3://bucket/base/" + name))
+ .setStorageConfigInfo(
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/" + name + "/"))
+ .build())
+ .build());
+ }
+
+ /** A catalog request selecting {@link #TEST_MECHANISM}, used at create time. */
+ private static Catalog testMechanismCatalog(String name) {
+ return PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(name)
+ .setProperties(new CatalogProperties("s3://bucket/base/" + name))
+ .setStorageConfigInfo(
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(TEST_MECHANISM)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/" + name + "/"))
+ .build())
+ .build();
+ }
+
+ /**
+ * A catalog created directly with {@link #TEST_MECHANISM}, installed at create time, in the realm
+ * the given headers target.
+ */
+ private static void createTestMechanismCatalog(
+ ManagementApi managementApi, Map<String, String> headers, String name) {
+ try (Response r =
+ managementApi
+ .request("v1/catalogs", Map.of(), Map.of(), headers)
+ .post(Entity.json(new CreateCatalogRequest(testMechanismCatalog(name))))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+ }
+ }
+
+ /** A catalog request selecting a mechanism the server never ships, used at create time. */
+ private static Catalog uninstalledMechanismCatalog(String name) {
+ return PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(name)
+ .setProperties(new CatalogProperties("s3://bucket/base/" + name))
+ .setStorageConfigInfo(
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(UNINSTALLED_MECHANISM)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/" + name + "/"))
+ .build())
+ .build();
+ }
+
+ /**
+ * Builds the request that would switch an existing catalog with an empty mechanism to a mechanism
+ * the server never ships. {@link ManagementApi#updateCatalog(Catalog, Map)} always reuses the
+ * existing storage config, so the switch needs a raw request carrying the new one.
+ */
+ private static UpdateCatalogRequest updateToUninstalledMechanismRequest(
+ ManagementApi managementApi, String name) {
+ Catalog fetched = managementApi.getCatalog(name);
+ return new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of("default-base-location", "s3://bucket/base/" + name),
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(UNINSTALLED_MECHANISM)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/" + name + "/"))
+ .build());
+ }
+
+ private static void createTable(CatalogApi catalogApi, String catalog, String ns, String table) {
+ CreateTableRequest request =
+ CreateTableRequest.builder()
+ .withName(table)
+ .withSchema(PolarisAuthzTestBase.SCHEMA)
+ .withLocation("s3://bucket/base/" + catalog + "/" + ns + "/" + table + "/")
+ .build();
+ try (Response r =
+ catalogApi
+ .request("v1/{cat}/namespaces/{ns}/tables", Map.of("cat", catalog, "ns", ns))
+ .post(Entity.json(request))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ }
+
+ private static void assertRefused(Response response, String expectedMessage) {
+ try (response) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.BAD_REQUEST.getStatusCode());
+ ErrorResponse body = ErrorResponseParser.fromJson(response.readEntity(String.class));
+ assertThat(body.message()).isEqualTo(expectedMessage);
+ }
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismDispatchTest.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismDispatchTest.java
new file mode 100644
index 0000000000..e1d2570382
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismDispatchTest.java
@@ -0,0 +1,174 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.auth.oauth2.AccessToken;
+import com.google.auth.oauth2.GoogleCredentials;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.config.RealmConfigImpl;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.entity.CatalogEntity;
+import org.apache.polaris.core.storage.PolarisStorageIntegration;
+import org.apache.polaris.core.storage.aws.AwsCredentialsStorageIntegration;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import software.amazon.awssdk.services.sts.StsClient;
+
+/** The provider's {@code case S3} checks the allowlist, then dispatches through the registry. */
+class S3CredentialVendingMechanismDispatchTest {
+
+ private static final RealmContext REALM = () -> "test-realm";
+
+ private static RealmConfig realmConfig(List<String> mechanisms) {
+ Map<String, Object> config = Map.of("SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS", mechanisms);
+ return new RealmConfigImpl((rc, name) -> config.get(name), REALM);
+ }
+
+ private static S3CredentialVendingMechanism testStsMechanism(RealmConfig realmConfig) {
+ return new StsCredentialVendingMechanism(
+ destination -> Mockito.mock(StsClient.class), Optional.empty(), null, realmConfig);
+ }
+
+ private static PolarisStorageIntegrationProviderImpl provider(
+ RealmConfig realmConfig, Map<String, S3CredentialVendingMechanism> mechanisms) {
+ return new PolarisStorageIntegrationProviderImpl(
+ new S3CredentialVendingMechanisms(mechanisms),
+ () -> GoogleCredentials.create(new AccessToken("abc", new Date())),
+ null,
+ realmConfig);
+ }
+
+ private static PolarisStorageIntegrationProviderImpl provider(RealmConfig realmConfig) {
+ return provider(realmConfig, Map.of("STS", testStsMechanism(realmConfig)));
+ }
+
+ private static CatalogEntity catalog(RealmConfig realmConfig, AwsStorageConfigInfo model) {
+ return new CatalogEntity.Builder()
+ .setName("c")
+ .addProperty(CatalogEntity.DEFAULT_BASE_LOCATION_KEY, "s3://bucket/base/")
+ .setStorageConfigurationInfo(realmConfig, model)
+ .build();
+ }
+
+ private static AwsStorageConfigInfo emptyMechanism() {
+ return AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/"))
+ .build();
+ }
+
+ private static AwsStorageConfigInfo uninstalled() {
+ return AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism("UNINSTALLED_MECHANISM")
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/"))
+ .build();
+ }
+
+ private static AwsStorageConfigInfo withMechanism(String mechanism) {
+ return AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(mechanism)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/"))
+ .build();
+ }
+
+ @Test
+ void stsAndAbsentMechanismDispatchToTheAwsIntegration() {
+ RealmConfig rc = realmConfig(List.of("STS"));
+ PolarisStorageIntegrationProviderImpl provider = provider(rc);
+ assertThat(provider.getStorageIntegration(List.of(catalog(rc, emptyMechanism()))))
+ .isInstanceOf(AwsCredentialsStorageIntegration.class);
+ assertThat(provider.getStorageIntegration(List.of(catalog(rc, withMechanism("STS")))))
+ .isInstanceOf(AwsCredentialsStorageIntegration.class);
+ }
+
+ @Test
+ void disallowedMechanismIsRejectedBeforeDispatch() {
+ RealmConfig rc = realmConfig(List.of("SECOND_MECHANISM"));
+ assertThatThrownBy(
+ () -> provider(rc).getStorageIntegration(List.of(catalog(rc, withMechanism("STS")))))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage("S3 credential vending mechanism STS is not enabled in this realm");
+ assertThat(provider(rc).getStorageIntegration(List.of(catalog(rc, emptyMechanism()))))
+ .isInstanceOf(AwsCredentialsStorageIntegration.class);
+ }
+
+ @Test
+ void allowlistedButUninstalledMechanismIsNotAvailable() {
+ RealmConfig rc = realmConfig(List.of("STS", "UNINSTALLED_MECHANISM"));
+ assertThatThrownBy(
+ () -> provider(rc).getStorageIntegration(List.of(catalog(rc, uninstalled()))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "S3 credential vending mechanism UNINSTALLED_MECHANISM is not available in this server");
+ }
+
+ @Test
+ void anInstalledThirdMechanismIsSelectedWhenAllowlisted() {
+ RealmConfig rc = realmConfig(List.of("STS", "TEST_MECHANISM"));
+ PolarisStorageIntegration expected = Mockito.mock(PolarisStorageIntegration.class);
+ AtomicInteger factoryCalls = new AtomicInteger();
+ S3CredentialVendingMechanism testMechanism =
+ storageConfig -> {
+ factoryCalls.incrementAndGet();
+ return expected;
+ };
+ PolarisStorageIntegrationProviderImpl provider =
+ provider(rc, Map.of("STS", testStsMechanism(rc), "TEST_MECHANISM", testMechanism));
+
+ assertThat(
+ provider.getStorageIntegration(List.of(catalog(rc, withMechanism("TEST_MECHANISM")))))
+ .isSameAs(expected);
+ assertThat(factoryCalls).hasValue(1);
+ }
+
+ @Test
+ void anInstalledThirdMechanismNotAllowlistedIsRefusedBeforeItsFactoryRuns() {
+ RealmConfig rc = realmConfig(List.of("STS"));
+ AtomicInteger factoryCalls = new AtomicInteger();
+ S3CredentialVendingMechanism testMechanism =
+ storageConfig -> {
+ factoryCalls.incrementAndGet();
+ return Mockito.mock(PolarisStorageIntegration.class);
+ };
+ PolarisStorageIntegrationProviderImpl provider =
+ provider(rc, Map.of("STS", testStsMechanism(rc), "TEST_MECHANISM", testMechanism));
+
+ assertThatThrownBy(
+ () ->
+ provider.getStorageIntegration(
+ List.of(catalog(rc, withMechanism("TEST_MECHANISM")))))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage("S3 credential vending mechanism TEST_MECHANISM is not enabled in this realm");
+ assertThat(factoryCalls).hasValue(0);
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismStsOverrideCdiTest.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismStsOverrideCdiTest.java
new file mode 100644
index 0000000000..4de902f641
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismStsOverrideCdiTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.client.Entity;
+import jakarta.ws.rs.core.Response;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.CatalogGrant;
+import org.apache.polaris.core.admin.model.CatalogPrivilege;
+import org.apache.polaris.core.admin.model.CatalogProperties;
+import org.apache.polaris.core.admin.model.GrantResource;
+import org.apache.polaris.core.admin.model.PolarisCatalog;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.entity.PolarisEntityConstants;
+import org.apache.polaris.core.storage.StorageAccessProperty;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
+import org.apache.polaris.service.admin.PolarisAuthzTestBase;
+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.service.it.ext.PolarisIntegrationTestExtension;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Proves that a server's enabled alternative for the STS identifier serves both an empty {@code
+ * credentialVendingMechanism} and an explicit {@code STS}. Runs under {@link
+ * RecordingStsMechanismProfile}, which enables {@link RecordingStsCredentialVendingMechanism} as a
+ * CDI alternative carrying {@code @Identifier(STS)}.
+ */
+@QuarkusTest
+@TestProfile(RecordingStsMechanismProfile.class)
+@ExtendWith(PolarisIntegrationTestExtension.class)
+class S3CredentialVendingMechanismStsOverrideCdiTest {
+
+ @Inject S3CredentialVendingMechanisms mechanisms;
+
+ @Inject
+ @Identifier(S3CredentialVendingMechanism.STS)
+ RecordingStsCredentialVendingMechanism override;
+
+ @Test
+ void anEmptyAndAnExplicitStsMechanismBothVendThroughTheEnabledAlternative(
+ PolarisApiEndpoints endpoints, ClientCredentials credentials) throws Exception {
+ override.clear();
+ assertThat(mechanisms.availableIds()).containsExactly("STS");
+ assertThat(mechanisms.require("STS"))
+ .isInstanceOf(RecordingStsCredentialVendingMechanism.class);
+
+ try (PolarisClient client = PolarisClient.polarisClient(endpoints)) {
+ String adminToken = client.obtainToken(credentials);
+ ManagementApi managementApi = client.managementApi(adminToken);
+ CatalogApi catalogApi = client.catalogApi(adminToken);
+
+ assertVendsThroughOverride(
+ managementApi, catalogApi, "cdi-sts-override-empty-cat", /* explicitSts= */ false);
+ assertVendsThroughOverride(
+ managementApi, catalogApi, "cdi-sts-override-explicit-cat", /* explicitSts= */ true);
+ }
+ }
+
+ private void assertVendsThroughOverride(
+ ManagementApi managementApi, CatalogApi catalogApi, String catalog, boolean explicitSts) {
+ override.clear();
+ managementApi.createCatalog(mechanismCatalog(catalog, explicitSts));
+ managementApi.addGrant(
+ catalog,
+ PolarisEntityConstants.getNameOfCatalogAdminRole(),
+ new CatalogGrant(CatalogPrivilege.CATALOG_MANAGE_CONTENT, GrantResource.TypeEnum.CATALOG));
+ catalogApi.createNamespace(catalog, "ns");
+ createTable(catalogApi, catalog, "ns", "t");
+
+ LoadTableResponse loaded =
+ catalogApi.loadTableWithAccessDelegation(
+ catalog, TableIdentifier.of(Namespace.of("ns"), "t"), null);
+ assertThat(loaded.credentials()).hasSize(1);
+ assertThat(loaded.credentials().get(0).config())
+ .containsEntry(
+ StorageAccessProperty.AWS_KEY_ID.getPropertyName(),
+ RecordingStsCredentialVendingMechanism.FAKE_KEY_FOR_TEST);
+ assertThat(override.calls()).isNotEmpty();
+ assertThat(override.calls())
+ .allSatisfy(
+ call ->
+ assertThat(call.storageConfig().getCredentialVendingMechanism())
+ .isEqualTo(explicitSts ? "STS" : null));
+ }
+
+ private static Catalog mechanismCatalog(String name, boolean explicitSts) {
+ AwsStorageConfigInfo.Builder storage =
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of("s3://bucket/base/" + name + "/"));
+ if (explicitSts) {
+ storage.setCredentialVendingMechanism("STS");
+ }
+ return PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(name)
+ .setProperties(new CatalogProperties("s3://bucket/base/" + name))
+ .setStorageConfigInfo(storage.build())
+ .build();
+ }
+
+ private static void createTable(CatalogApi catalogApi, String catalog, String ns, String table) {
+ CreateTableRequest request =
+ CreateTableRequest.builder()
+ .withName(table)
+ .withSchema(PolarisAuthzTestBase.SCHEMA)
+ .withLocation("s3://bucket/base/" + catalog + "/" + ns + "/" + table + "/")
+ .build();
+ try (Response r =
+ catalogApi
+ .request("v1/{cat}/namespaces/{ns}/tables", Map.of("cat", catalog, "ns", ns))
+ .post(Entity.json(request))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismThirdMechanismCdiTest.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismThirdMechanismCdiTest.java
new file mode 100644
index 0000000000..924723b16b
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismThirdMechanismCdiTest.java
@@ -0,0 +1,243 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.client.Entity;
+import jakarta.ws.rs.core.Response;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.responses.ErrorResponse;
+import org.apache.iceberg.rest.responses.ErrorResponseParser;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.CatalogGrant;
+import org.apache.polaris.core.admin.model.CatalogPrivilege;
+import org.apache.polaris.core.admin.model.CatalogProperties;
+import org.apache.polaris.core.admin.model.CreateCatalogRequest;
+import org.apache.polaris.core.admin.model.GrantResource;
+import org.apache.polaris.core.admin.model.PolarisCatalog;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.admin.model.UpdateCatalogRequest;
+import org.apache.polaris.core.entity.PolarisEntityConstants;
+import org.apache.polaris.core.storage.StorageAccessProperty;
+import org.apache.polaris.service.admin.PolarisAuthzTestBase;
+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.service.it.ext.PolarisIntegrationTestExtension;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * A third {@code S3CredentialVendingMechanism} the server does not ship: proves the registry and
+ * the gates generalize past {@code STS}. Runs under {@link ThirdMechanismProfile}, which enables
+ * {@link RecordingThirdCredentialVendingMechanism} as a CDI alternative and allowlists it for one
+ * realm but not a second.
+ *
+ * <p>This is a separate top-level class, not a {@code @Nested} class inside {@link
+ * S3CredentialVendingMechanismCdiTest}, because Quarkus rejects {@code @TestProfile} on
+ * {@code @Nested} test classes ({@code io.quarkus.test.junit.QuarkusTestExtension}: "@Nested tests
+ * may not contain @TestProfile annotations"), and the two scenarios need different application
+ * instances regardless: {@code getEnabledAlternatives()} is profile-wide, and {@link
+ * S3CredentialVendingMechanismCdiTest} asserts {@code availableIds()} is exactly {@code {STS}} in
+ * its own application instance, which this test's third mechanism would otherwise widen.
+ */
+@QuarkusTest
+@TestProfile(ThirdMechanismProfile.class)
+@ExtendWith(PolarisIntegrationTestExtension.class)
+class S3CredentialVendingMechanismThirdMechanismCdiTest {
+
+ @Inject S3CredentialVendingMechanisms mechanisms;
+
+ @Inject
+ @Identifier(RecordingThirdCredentialVendingMechanism.ID)
+ RecordingThirdCredentialVendingMechanism testMechanism;
+
+ @Test
+ void anInstalledThirdMechanismVendsWhenAllowlistedAndIsRefusedWithoutDispatchWhenNot(
+ PolarisApiEndpoints endpoints, ClientCredentials credentials) throws Exception {
+ testMechanism.clear();
+ assertThat(mechanisms.availableIds())
+ .containsExactly("STS", RecordingThirdCredentialVendingMechanism.ID);
+ try (PolarisClient client = PolarisClient.polarisClient(endpoints)) {
+ String adminToken = client.obtainToken(credentials);
+ ManagementApi managementApi = client.managementApi(adminToken);
+ CatalogApi catalogApi = client.catalogApi(adminToken);
+
+ String catalog = "cdi-test-mech-cat";
+ managementApi.createCatalog(testMechanismCatalog(catalog));
+ assertThat(testMechanism.validations()).hasSize(1);
+ assertThat(testMechanism.validations().get(0).current()).isNull();
+ assertThat(testMechanism.validations().get(0).updated().getCredentialVendingMechanism())
+ .isEqualTo(RecordingThirdCredentialVendingMechanism.ID);
+
+ // A config the mechanism itself refuses at validate() time: CatalogEntity accepts it (the
+ // base location sits inside the one allowed location) and the refusal comes from validate().
+ String refusedBase = "s3://bucket/base/refused/cdi-refused-cat";
+ try (Response refused =
+ managementApi
+ .request("v1/catalogs")
+ .post(
+ Entity.json(
+ new CreateCatalogRequest(
+ testMechanismCatalog("cdi-refused-cat", refusedBase))))) {
+ assertRefused(refused, "TEST_MECHANISM refuses the allowed location " + refusedBase + "/");
+ }
+
+ // The catalog's auto-granted catalog_admin role (assigned to service_admin at creation)
+ // carries CATALOG_MANAGE_ACCESS and CATALOG_MANAGE_METADATA only; loadTable with vended
+ // credentials needs CATALOG_MANAGE_CONTENT too, the same grant ManagementApi.makeAdmin gives
+ // a caller-supplied catalog role.
+ managementApi.addGrant(
+ catalog,
+ PolarisEntityConstants.getNameOfCatalogAdminRole(),
+ new CatalogGrant(
+ CatalogPrivilege.CATALOG_MANAGE_CONTENT, GrantResource.TypeEnum.CATALOG));
+ catalogApi.createNamespace(catalog, "ns");
+ createTable(catalogApi, catalog, "ns", "t");
+
+ LoadTableResponse loaded =
+ catalogApi.loadTableWithAccessDelegation(
+ catalog, TableIdentifier.of(Namespace.of("ns"), "t"), null);
+ assertThat(loaded.credentials()).hasSize(1);
+ assertThat(loaded.credentials().get(0).config())
+ .containsEntry(
+ StorageAccessProperty.AWS_KEY_ID.getPropertyName(),
+ RecordingThirdCredentialVendingMechanism.FAKE_KEY_FOR_TEST)
+ .containsEntry(
+ StorageAccessProperty.AWS_SECRET_KEY.getPropertyName(),
+ RecordingThirdCredentialVendingMechanism.FAKE_SECRET_FOR_TEST)
+ .containsEntry(
+ StorageAccessProperty.AWS_TOKEN.getPropertyName(),
+ RecordingThirdCredentialVendingMechanism.FAKE_TOKEN_FOR_TEST);
+ // The end-to-end create-table-then-loadTable-with-delegation flow dispatches to the
+ // mechanism more than once (once per storage action grouping FileIO and the credential
+ // provider resolve internally); that fan-out is pre-existing StorageAccessConfigProvider
+ // behaviour, unchanged by this feature, and orthogonal to what this test proves. What
+ // matters here: the mechanism was reached at all, and every call it saw carried this
+ // catalog's own config.
+ assertThat(testMechanism.calls()).isNotEmpty();
+ int callsAfterVending = testMechanism.calls().size();
+ assertThat(testMechanism.calls())
+ .allSatisfy(
+ call -> {
+ assertThat(call.storageConfig().getCredentialVendingMechanism())
+ .isEqualTo(RecordingThirdCredentialVendingMechanism.ID);
+ assertThat(call.storageConfig().getAllowedLocations())
+ .containsExactly("s3://bucket/base/" + catalog + "/");
+ });
+
+ // A realm where TEST_MECHANISM is not allowlisted: creation itself is refused with "not
+ // enabled", at PolarisAdminService's allowlist-only gate, and the mechanism is never
+ // dispatched, so the recorder gains no further call.
+ Map<String, String> killSwitchRealmHeaders =
+ Map.of(
+ "Authorization",
+ "Bearer " + adminToken,
+ "Polaris-Realm",
+ ThirdMechanismProfile.KILL_SWITCH_REALM);
+ try (Response refused =
+ managementApi
+ .request("v1/catalogs", Map.of(), Map.of(), killSwitchRealmHeaders)
+ .post(Entity.json(new CreateCatalogRequest(testMechanismCatalog("cdi-kill-cat"))))) {
+ assertRefused(
+ refused, "S3 credential vending mechanism TEST_MECHANISM is not enabled in this realm");
+ }
+ assertThat(testMechanism.calls()).hasSize(callsAfterVending);
+
+ // Updating the first catalog with a second allowed location validates again, this time
+ // with the stored config as current().
+ Catalog fetched = managementApi.getCatalog(catalog);
+ int validationsBeforeUpdate = testMechanism.validations().size();
+ UpdateCatalogRequest addLocation =
+ new UpdateCatalogRequest(
+ fetched.getEntityVersion(),
+ Map.of(),
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(RecordingThirdCredentialVendingMechanism.ID)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(
+ List.of(
+ "s3://bucket/base/" + catalog + "/",
+ "s3://bucket/base/" + catalog + "-extra/"))
+ .build());
+ try (Response r =
+ managementApi
+ .request("v1/catalogs/{name}", Map.of("name", catalog))
+ .put(Entity.json(addLocation))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ assertThat(testMechanism.validations()).hasSize(validationsBeforeUpdate + 1);
+ assertThat(testMechanism.validations().get(validationsBeforeUpdate).current()).isNotNull();
+ }
+ }
+
+ private static Catalog testMechanismCatalog(String name) {
+ return testMechanismCatalog(name, "s3://bucket/base/" + name);
+ }
+
+ private static Catalog testMechanismCatalog(String name, String basePath) {
+ return PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(name)
+ .setProperties(new CatalogProperties(basePath))
+ .setStorageConfigInfo(
+ AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+ .setCredentialVendingMechanism(RecordingThirdCredentialVendingMechanism.ID)
+ .setRoleArn("arn:aws:iam::123456789012:role/r")
+ .setAllowedLocations(List.of(basePath + "/"))
+ .build())
+ .build();
+ }
+
+ private static void createTable(CatalogApi catalogApi, String catalog, String ns, String table) {
+ CreateTableRequest request =
+ CreateTableRequest.builder()
+ .withName(table)
+ .withSchema(PolarisAuthzTestBase.SCHEMA)
+ .withLocation("s3://bucket/base/" + catalog + "/" + ns + "/" + table + "/")
+ .build();
+ try (Response r =
+ catalogApi
+ .request("v1/{cat}/namespaces/{ns}/tables", Map.of("cat", catalog, "ns", ns))
+ .post(Entity.json(request))) {
+ assertThat(r.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ }
+
+ private static void assertRefused(Response response, String expectedMessage) {
+ try (response) {
+ assertThat(response.getStatus()).isEqualTo(Response.Status.BAD_REQUEST.getStatusCode());
+ ErrorResponse body = ErrorResponseParser.fromJson(response.readEntity(String.class));
+ assertThat(body.message()).isEqualTo(expectedMessage);
+ }
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismsTest.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismsTest.java
new file mode 100644
index 0000000000..49f11320cc
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/S3CredentialVendingMechanismsTest.java
@@ -0,0 +1,198 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.enterprise.inject.spi.Bean;
+import jakarta.enterprise.inject.spi.BeanManager;
+import java.lang.annotation.Annotation;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
+import org.junit.jupiter.api.Test;
+
+/**
+ * {@link S3CredentialVendingMechanisms} discovers identifiers from bean metadata and resolves each
+ * identifier once, at construction. No CDI container runs here: the {@link BeanManager} and the
+ * {@link Instance} are Mockito fakes that answer only the calls the registry makes.
+ */
+class S3CredentialVendingMechanismsTest {
+
+ private static final class StsBean {}
+
+ private static final class OtherBean {}
+
+ private static final class OverridingStsBean {}
+
+ private static final class SecondBean {}
+
+ private final BeanManager beanManager = mock(BeanManager.class);
+
+ @SuppressWarnings("unchecked")
+ private final Instance<S3CredentialVendingMechanism> candidates = mock(Instance.class);
+
+ @Test
+ void identifiersComeFromEveryBeanAndResolutionUsesTheSelectedBean() {
+ S3CredentialVendingMechanism override = mock(S3CredentialVendingMechanism.class);
+ S3CredentialVendingMechanism other = mock(S3CredentialVendingMechanism.class);
+ beans(
+ bean(StsBean.class, Identifier.Literal.of("STS")),
+ bean(OverridingStsBean.class, Identifier.Literal.of("STS")),
+ bean(OtherBean.class, Identifier.Literal.of("OTHER")));
+ resolves("STS", override);
+ resolves("OTHER", other);
+
+ S3CredentialVendingMechanisms registry =
+ new S3CredentialVendingMechanisms(candidates, beanManager);
+
+ assertThat(registry.availableIds()).containsExactly("OTHER", "STS");
+ assertThat(registry.require("STS")).isSameAs(override);
+ assertThat(registry.require("OTHER")).isSameAs(other);
+ }
+
+ @Test
+ void aBeanWithNoIdentifierAbortsStartup() {
+ beans(bean(StsBean.class));
+
+ assertThatThrownBy(() -> new S3CredentialVendingMechanisms(candidates, beanManager))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "S3 credential vending mechanism bean "
+ + StsBean.class.getName()
+ + " has no @Identifier");
+ }
+
+ @Test
+ void twoBeansSharingOneIdentifierAbortStartup() {
+ beans(
+ bean(StsBean.class, Identifier.Literal.of("DUP")),
+ bean(SecondBean.class, Identifier.Literal.of("DUP")));
+ ambiguous("DUP");
+
+ assertThatThrownBy(() -> new S3CredentialVendingMechanisms(candidates, beanManager))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Two S3 credential vending mechanisms share the identifier DUP");
+ }
+
+ @Test
+ void anIdentifierThatResolvesToNoBeanAbortsStartup() {
+ beans(bean(StsBean.class, Identifier.Literal.of("GHOST")));
+ unsatisfied("GHOST");
+
+ assertThatThrownBy(() -> new S3CredentialVendingMechanisms(candidates, beanManager))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("S3 credential vending mechanism GHOST resolves to no enabled bean");
+ }
+
+ @Test
+ void aBeanThatFailsToConstructAbortsStartup() {
+ beans(bean(StsBean.class, Identifier.Literal.of("STS")));
+ failsToConstruct("STS", new IllegalStateException("no STS client"));
+
+ assertThatThrownBy(() -> new S3CredentialVendingMechanisms(candidates, beanManager))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("no STS client");
+ }
+
+ @Test
+ void requireOnAMissingIdentifierIsAnIllegalArgument() {
+ beans(bean(StsBean.class, Identifier.Literal.of("STS")));
+ resolves("STS", mock(S3CredentialVendingMechanism.class));
+ S3CredentialVendingMechanisms registry =
+ new S3CredentialVendingMechanisms(candidates, beanManager);
+
+ assertThat(registry.isAvailable("NOPE")).isFalse();
+ assertThatThrownBy(() -> registry.require("NOPE"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("S3 credential vending mechanism NOPE is not available in this server");
+ }
+
+ @Test
+ void theTestMapIsReadOnEveryCall() {
+ Map<String, S3CredentialVendingMechanism> live = new HashMap<>();
+ live.put("STS", mock(S3CredentialVendingMechanism.class));
+ S3CredentialVendingMechanisms registry = new S3CredentialVendingMechanisms(live);
+ assertThat(registry.availableIds()).containsExactly("STS");
+
+ live.put("LATER", mock(S3CredentialVendingMechanism.class));
+ assertThat(registry.availableIds()).containsExactly("LATER", "STS");
+ assertThat(registry.isAvailable("LATER")).isTrue();
+
+ live.remove("STS");
+ assertThat(registry.isAvailable("STS")).isFalse();
+ assertThatThrownBy(() -> registry.require("STS"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("S3 credential vending mechanism STS is not available in this server");
+ }
+
+ private void beans(Bean<?>... beans) {
+ doReturn(Set.of(beans))
+ .when(beanManager)
+ .getBeans(S3CredentialVendingMechanism.class, Any.Literal.INSTANCE);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Bean<?> bean(Class<?> beanClass, Annotation... qualifiers) {
+ Bean<S3CredentialVendingMechanism> bean = mock(Bean.class);
+ doReturn(beanClass).when(bean).getBeanClass();
+ doReturn(Set.of(qualifiers)).when(bean).getQualifiers();
+ return bean;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Instance<S3CredentialVendingMechanism> selection(String id) {
+ Instance<S3CredentialVendingMechanism> selected = mock(Instance.class);
+ doReturn(selected).when(candidates).select(Identifier.Literal.of(id));
+ return selected;
+ }
+
+ private void resolves(String id, S3CredentialVendingMechanism mechanism) {
+ Instance<S3CredentialVendingMechanism> selected = selection(id);
+ doReturn(false).when(selected).isAmbiguous();
+ doReturn(false).when(selected).isUnsatisfied();
+ doReturn(mechanism).when(selected).get();
+ }
+
+ private void ambiguous(String id) {
+ Instance<S3CredentialVendingMechanism> selected = selection(id);
+ doReturn(true).when(selected).isAmbiguous();
+ }
+
+ private void unsatisfied(String id) {
+ Instance<S3CredentialVendingMechanism> selected = selection(id);
+ doReturn(false).when(selected).isAmbiguous();
+ doReturn(true).when(selected).isUnsatisfied();
+ }
+
+ private void failsToConstruct(String id, RuntimeException failure) {
+ Instance<S3CredentialVendingMechanism> selected = selection(id);
+ doReturn(false).when(selected).isAmbiguous();
+ doReturn(false).when(selected).isUnsatisfied();
+ doThrow(failure).when(selected).get();
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/StsCredentialVendingMechanismTest.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/StsCredentialVendingMechanismTest.java
new file mode 100644
index 0000000000..ef8dd8e652
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/StsCredentialVendingMechanismTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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;
+
+import static org.apache.polaris.core.config.FeatureConfiguration.RESOLVE_CREDENTIALS_BY_STORAGE_NAME;
+import static org.apache.polaris.core.config.FeatureConfiguration.STORAGE_CREDENTIAL_DURATION_SECONDS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.config.RealmConfigImpl;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.storage.CredentialVendingContext;
+import org.apache.polaris.core.storage.LocationGrant;
+import org.apache.polaris.core.storage.PolarisStorageActions;
+import org.apache.polaris.core.storage.PolarisStorageIntegration;
+import org.apache.polaris.core.storage.StorageAccessConfig;
+import org.apache.polaris.core.storage.StorageAccessProperty;
+import org.apache.polaris.core.storage.aws.AwsCredentialsStorageIntegration;
+import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+import org.apache.polaris.core.storage.aws.StsClientProvider;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
+
+/**
+ * {@link StsCredentialVendingMechanism} moved {@code AwsCredentialsStorageIntegration} construction
+ * out of the provider unchanged; these tests prove that move preserved the existing behaviour:
+ * per-storage-name credential resolution, {@code stsUnavailable}, and per-realm-config credential
+ * duration. The mock {@link StorageConfiguration} is never delegated to its default methods: every
+ * {@code stsCredentials(...)} overload used here is stubbed directly. The CDI-level discovery and
+ * gating behaviour lives in {@link S3CredentialVendingMechanismCdiTest}.
+ */
+class StsCredentialVendingMechanismTest {
+
+ private static final RealmContext REALM = () -> "test-realm";
+ private static final String ROLE_ARN = "arn:aws:iam::123456789012:role/r";
+ private static final String LOCATION = "s3://bucket/base/";
+
+ private static RealmConfig realmConfig(Map<String, Object> overrides) {
+ return new RealmConfigImpl((rc, name) -> overrides.get(name), REALM);
+ }
+
+ private static AwsStorageConfigurationInfo storageConfig() {
+ return AwsStorageConfigurationInfo.builder()
+ .roleARN(ROLE_ARN)
+ .addAllowedLocation(LOCATION)
+ .build();
+ }
+
+ private static AwsStorageConfigurationInfo storageConfigWithName(String storageName) {
+ return AwsStorageConfigurationInfo.builder()
+ .roleARN(ROLE_ARN)
+ .storageName(storageName)
+ .addAllowedLocation(LOCATION)
+ .build();
+ }
+
+ private static AssumeRoleResponse assumeRoleResponse() {
+ return AssumeRoleResponse.builder()
+ .credentials(
+ Credentials.builder()
+ .accessKeyId("accessKey")
+ .secretAccessKey("secretKey")
+ .sessionToken("sess")
+ .build())
+ .build();
+ }
+
+ private static List<LocationGrant> grants() {
+ return List.of(new LocationGrant(Set.of(LOCATION + "t/"), Set.of(PolarisStorageActions.READ)));
+ }
+
+ @Test
+ void resolveCredentialsByStorageNameTrueAsksForTheNamedStorage() {
+ StorageConfiguration storageConfiguration = mock(StorageConfiguration.class);
+ AwsCredentialsProvider namedCredentials =
+ StaticCredentialsProvider.create(AwsBasicCredentials.create("ak", "sk"));
+ when(storageConfiguration.stsCredentials("named-storage")).thenReturn(namedCredentials);
+
+ StsClient stsClient = mock(StsClient.class);
+ ArgumentCaptor<AssumeRoleRequest> captor = ArgumentCaptor.forClass(AssumeRoleRequest.class);
+ when(stsClient.assumeRole(captor.capture())).thenReturn(assumeRoleResponse());
+ StsClientProvider stsClientProvider = destination -> stsClient;
+
+ RealmConfig realmConfig = realmConfig(Map.of(RESOLVE_CREDENTIALS_BY_STORAGE_NAME.key(), true));
+ StsCredentialVendingMechanism mechanism =
+ new StsCredentialVendingMechanism(
+ storageConfiguration, stsClientProvider, null, realmConfig);
+ PolarisStorageIntegration integration =
+ mechanism.integrationFor(storageConfigWithName("named-storage"));
+
+ integration.getStorageAccessConfig(
+ grants(), Optional.empty(), CredentialVendingContext.empty());
+
+ verify(storageConfiguration).stsCredentials("named-storage");
+ verify(storageConfiguration, never()).stsCredentials();
+ assertThat(captor.getValue().overrideConfiguration()).isPresent();
+ }
+
+ @Test
+ void resolveCredentialsByStorageNameFalseAsksForTheDefault() {
+ StorageConfiguration storageConfiguration = mock(StorageConfiguration.class);
+ when(storageConfiguration.stsCredentials())
+ .thenReturn(StaticCredentialsProvider.create(AwsBasicCredentials.create("ak", "sk")));
+
+ StsClient stsClient = mock(StsClient.class);
+ when(stsClient.assumeRole(any(AssumeRoleRequest.class))).thenReturn(assumeRoleResponse());
+ StsClientProvider stsClientProvider = destination -> stsClient;
+
+ RealmConfig realmConfig = realmConfig(Map.of(RESOLVE_CREDENTIALS_BY_STORAGE_NAME.key(), false));
+ StsCredentialVendingMechanism mechanism =
+ new StsCredentialVendingMechanism(
+ storageConfiguration, stsClientProvider, null, realmConfig);
+ PolarisStorageIntegration integration =
+ mechanism.integrationFor(storageConfigWithName("named-storage"));
+
+ integration.getStorageAccessConfig(
+ grants(), Optional.empty(), CredentialVendingContext.empty());
+
+ verify(storageConfiguration).stsCredentials();
+ verify(storageConfiguration, never()).stsCredentials(anyString());
+ }
+
+ @Test
+ void stsUnavailableVendsNoCredentialsAndNeverCallsSts() {
+ StorageConfiguration storageConfiguration = mock(StorageConfiguration.class);
+ StsClient stsClient = mock(StsClient.class);
+ StsClientProvider stsClientProvider = destination -> stsClient;
+
+ RealmConfig realmConfig = realmConfig(Map.of());
+ StsCredentialVendingMechanism mechanism =
+ new StsCredentialVendingMechanism(
+ storageConfiguration, stsClientProvider, null, realmConfig);
+ AwsStorageConfigurationInfo config =
+ AwsStorageConfigurationInfo.builder()
+ .roleARN(ROLE_ARN)
+ .addAllowedLocation(LOCATION)
+ .stsUnavailable(true)
+ .build();
+ PolarisStorageIntegration integration = mechanism.integrationFor(config);
+
+ StorageAccessConfig accessConfig =
+ integration.getStorageAccessConfig(
+ grants(), Optional.empty(), CredentialVendingContext.empty());
+
+ verify(stsClient, never()).assumeRole(any(AssumeRoleRequest.class));
+ assertThat(accessConfig.credentials())
+ .doesNotContainKey(StorageAccessProperty.AWS_KEY_ID.getPropertyName())
+ .doesNotContainKey(StorageAccessProperty.AWS_SECRET_KEY.getPropertyName())
+ .doesNotContainKey(StorageAccessProperty.AWS_TOKEN.getPropertyName());
+ }
+
+ @Test
+ void twoRealmConfigsProduceIntegrationsThatReadTheirOwnCredentialDuration() {
+ StorageConfiguration storageConfiguration = mock(StorageConfiguration.class);
+ when(storageConfiguration.stsCredentials())
+ .thenReturn(StaticCredentialsProvider.create(AwsBasicCredentials.create("ak", "sk")));
+
+ StsClient stsClient = mock(StsClient.class);
+ ArgumentCaptor<AssumeRoleRequest> captor = ArgumentCaptor.forClass(AssumeRoleRequest.class);
+ when(stsClient.assumeRole(captor.capture())).thenReturn(assumeRoleResponse());
+ StsClientProvider stsClientProvider = destination -> stsClient;
+
+ AwsStorageConfigurationInfo config = storageConfig();
+
+ RealmConfig shortDuration = realmConfig(Map.of(STORAGE_CREDENTIAL_DURATION_SECONDS.key(), 900));
+ RealmConfig longDuration = realmConfig(Map.of(STORAGE_CREDENTIAL_DURATION_SECONDS.key(), 3600));
+ StsCredentialVendingMechanism shortMechanism =
+ new StsCredentialVendingMechanism(
+ storageConfiguration, stsClientProvider, null, shortDuration);
+ StsCredentialVendingMechanism longMechanism =
+ new StsCredentialVendingMechanism(
+ storageConfiguration, stsClientProvider, null, longDuration);
+
+ PolarisStorageIntegration shortIntegration = shortMechanism.integrationFor(config);
+ shortIntegration.getStorageAccessConfig(
+ grants(), Optional.empty(), CredentialVendingContext.empty());
+ assertThat(captor.getValue().durationSeconds()).isEqualTo(900);
+
+ PolarisStorageIntegration longIntegration = longMechanism.integrationFor(config);
+ longIntegration.getStorageAccessConfig(
+ grants(), Optional.empty(), CredentialVendingContext.empty());
+ assertThat(captor.getValue().durationSeconds()).isEqualTo(3600);
+ }
+
+ @Test
+ void theStsMechanismBuildsAnAwsCredentialsIntegration() {
+ StsClientProvider stsClientProvider = destination -> mock(StsClient.class);
+ AwsStorageConfigurationInfo config = storageConfig();
+ RealmConfig realmConfig = realmConfig(Map.of());
+
+ assertThat(
+ new StsCredentialVendingMechanism(
+ stsClientProvider, Optional.empty(), null, realmConfig)
+ .integrationFor(config))
+ .isInstanceOf(AwsCredentialsStorageIntegration.class);
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/TestInMemoryFileIOFactory.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/TestInMemoryFileIOFactory.java
new file mode 100644
index 0000000000..03476fd027
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/TestInMemoryFileIOFactory.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.storage;
+
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.RequestScoped;
+import java.util.Map;
+import org.apache.iceberg.inmemory.InMemoryFileIO;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.storage.StorageAccessConfig;
+import org.apache.polaris.service.catalog.io.FileIOFactory;
+import org.jspecify.annotations.NonNull;
+
+/**
+ * A {@code FileIO} factory selected by {@code polaris.file-io.type=test-in-memory}: every table
+ * write and read in these CDI tests goes through {@link InMemoryFileIO}, never a real cloud
+ * endpoint, keeping the S3 credential vending mechanism tests hermetic. Selected through the same
+ * {@code @Identifier}-and-config mechanism as {@code DefaultFileIOFactory} ({@code
+ * ServiceProducers.fileIOFactory}), so no CDI alternative is needed: {@code
+ * CatalogProperties.FILE_IO_IMPL} is never set on any catalog or request, so {@code
+ * IcebergPropertiesValidation.determineFileIOClassName}'s insecure-storage-type check (the one that
+ * would otherwise require {@code ALLOW_INSECURE_STORAGE_TYPES}, a severe readiness issue) never
+ * runs; the storage config still reports the real, safe {@code S3FileIO} class name, but this
+ * factory substitutes {@link InMemoryFileIO} underneath it regardless.
+ */
+@RequestScoped
+@Identifier("test-in-memory")
+public class TestInMemoryFileIOFactory implements FileIOFactory {
+
+ @Override
+ public FileIO loadFileIO(
+ @NonNull StorageAccessConfig storageAccessConfig,
+ @NonNull String ioImplClassName,
+ @NonNull Map<String, String> properties) {
+ return new InMemoryFileIO();
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/storage/ThirdMechanismProfile.java b/runtime/service/src/test/java/org/apache/polaris/service/storage/ThirdMechanismProfile.java
new file mode 100644
index 0000000000..2e7aa76f32
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/storage/ThirdMechanismProfile.java
@@ -0,0 +1,69 @@
+/*
+ * 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;
+
+import io.quarkus.test.junit.QuarkusTestProfile;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Installs {@link RecordingThirdCredentialVendingMechanism} as a CDI alternative and allowlists it
+ * for one realm ({@code POLARIS}) but not a second ({@code POLARIS2}): proves that an installed,
+ * non-STS mechanism vends when allowlisted and is refused, without ever being invoked, when it is
+ * not.
+ */
+public class ThirdMechanismProfile implements QuarkusTestProfile {
+
+ /**
+ * The realm the test client targets by default; the positive path never sends an explicit realm
+ * header, it just relies on this being the client's default realm. Named here only to build the
+ * realm list below.
+ */
+ public static final String ALLOWLISTED_REALM = "POLARIS";
+
+ public static final String KILL_SWITCH_REALM = "POLARIS2";
+
+ @Override
+ public Set<Class<?>> getEnabledAlternatives() {
+ return Set.of(RecordingThirdCredentialVendingMechanism.class);
+ }
+
+ @Override
+ public Map<String, String> getConfigOverrides() {
+ return Map.of(
+ "polaris.realm-context.realms",
+ ALLOWLISTED_REALM + "," + KILL_SWITCH_REALM,
+ "polaris.file-io.type",
+ "test-in-memory",
+ "polaris.features.\"SUPPORTED_CATALOG_STORAGE_TYPES\"",
+ "[\"S3\"]",
+ "polaris.features.\"SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS\"",
+ "[\"STS\",\"TEST_MECHANISM\"]",
+ "polaris.features.realm-overrides.\""
+ + KILL_SWITCH_REALM
+ + "\".\"SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS\"",
+ "[\"STS\"]",
+ "polaris.event-listener.type",
+ "test",
+ "polaris.authentication.token-broker.type",
+ "symmetric-key",
+ "polaris.authentication.token-broker.symmetric-key.secret",
+ "secret");
+ }
+}
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 da8178efc7..584200344c 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
@@ -32,6 +32,7 @@ import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.Date;
+import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -70,6 +71,9 @@ import org.apache.polaris.core.persistence.resolver.Resolver;
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.StorageAccessConfig;
+import org.apache.polaris.core.storage.StorageAccessProperty;
+import org.apache.polaris.core.storage.aws.S3CredentialVendingMechanism;
import org.apache.polaris.core.storage.cache.StorageCredentialCache;
import org.apache.polaris.core.storage.cache.StorageCredentialCacheConfig;
import org.apache.polaris.service.admin.PolarisAdminService;
@@ -118,6 +122,8 @@ import org.apache.polaris.service.identity.provider.DefaultServiceIdentityProvid
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.S3CredentialVendingMechanisms;
+import org.apache.polaris.service.storage.StsCredentialVendingMechanism;
import org.apache.polaris.service.task.TaskExecutor;
import org.mockito.Mockito;
import software.amazon.awssdk.services.sts.StsClient;
@@ -148,6 +154,8 @@ public record TestServices(
PolarisEventDispatcher polarisEventDispatcher,
PolarisEventMetadataFactory eventMetadataFactory,
StorageAccessConfigProvider storageAccessConfigProvider,
+ Map<String, S3CredentialVendingMechanism> installedMechanisms,
+ S3CredentialVendingMechanisms vendingMechanisms,
IdempotencyRequestContext idempotencyRequestContext) {
public PolarisCatalogsApi catalogsApi() {
@@ -183,6 +191,7 @@ public record TestServices(
private RealmContext realmContext = TEST_REALM;
private Map<String, Object> config = Map.of();
private StsClient stsClient;
+ private Map<String, S3CredentialVendingMechanism> additionalVendingMechanisms = Map.of();
private boolean useEventDelegator = false;
private Supplier<FileIOFactory> fileIOFactorySupplier = MeasuredFileIOFactory::new;
private UnaryOperator<PolarisMetaStoreManager> metaStoreManagerDecorator =
@@ -236,6 +245,17 @@ public record TestServices(
return this;
}
+ /**
+ * Mechanisms installed in addition to the defaults; an entry with a default identifier replaces
+ * the default. The map behind the registry stays mutable and is exposed as {@code
+ * installedMechanisms()}, so a test can uninstall a mechanism after a catalog selected it.
+ */
+ public Builder additionalVendingMechanisms(
+ Map<String, S3CredentialVendingMechanism> additionalVendingMechanisms) {
+ this.additionalVendingMechanisms = additionalVendingMechanisms;
+ return this;
+ }
+
public Builder withEventDelegator(boolean useEventDelegator) {
this.useEventDelegator = useEventDelegator;
return this;
@@ -266,10 +286,18 @@ public record TestServices(
RealmConfig realmConfig = new RealmConfigImpl(configurationSource, realmContext);
+ Map<String, S3CredentialVendingMechanism> vendingMechanismsMap = new HashMap<>();
+ vendingMechanismsMap.put(
+ S3CredentialVendingMechanism.STS,
+ new StsCredentialVendingMechanism(
+ (destination) -> stsClient, Optional.empty(), storageCredentialCache, realmConfig));
+ vendingMechanismsMap.putAll(additionalVendingMechanisms);
+ S3CredentialVendingMechanisms vendingMechanisms =
+ new S3CredentialVendingMechanisms(vendingMechanismsMap);
+
PolarisStorageIntegrationProviderImpl storageIntegrationProvider =
new PolarisStorageIntegrationProviderImpl(
- (destination) -> stsClient,
- Optional.empty(),
+ vendingMechanisms,
() -> GoogleCredentials.create(new AccessToken(GCP_ACCESS_TOKEN, new Date())),
storageCredentialCache,
realmConfig);
@@ -564,7 +592,8 @@ public record TestServices(
serviceIdentityProvider,
principal,
authorizer,
- reservedProperties);
+ reservedProperties,
+ vendingMechanisms);
return new PolarisCatalogsApi(
new PolarisServiceImpl(
realmConfig, reservedProperties, adminService, serviceIdentityProvider));
@@ -593,6 +622,8 @@ public record TestServices(
polarisEventDispatcher,
eventMetadataFactory,
storageAccessConfigProvider,
+ vendingMechanismsMap,
+ vendingMechanisms,
idempotencyRequestContext);
}
}
@@ -615,4 +646,20 @@ public record TestServices(
metaStoreManagerFactory.getOrCreateSession(realmContext),
configurationSource);
}
+
+ /**
+ * A mechanism that vends a fixed fake credential triple: for tests that need an installed second
+ * mechanism. A Mockito mock of the integration would return null from getStorageAccessConfig and
+ * break table creation whenever credential subscoping is not skipped.
+ */
+ public static S3CredentialVendingMechanism fakeMechanism() {
+ return storageConfig ->
+ (grants, refreshEndpoint, context) ->
+ StorageAccessConfig.builder()
+ .putCredential(StorageAccessProperty.AWS_KEY_ID.getPropertyName(), "FAKE_KEY")
+ .putCredential(
+ StorageAccessProperty.AWS_SECRET_KEY.getPropertyName(), "FAKE_SECRET")
+ .putCredential(StorageAccessProperty.AWS_TOKEN.getPropertyName(), "FAKE_TOKEN")
+ .build();
+ }
}
diff --git a/site/content/in-dev/unreleased/configuration/config-sections/flags-polaris_features.md b/site/content/in-dev/unreleased/configuration/config-sections/flags-polaris_features.md
index 741d062112..929c2e8886 100644
--- a/site/content/in-dev/unreleased/configuration/config-sections/flags-polaris_features.md
+++ b/site/content/in-dev/unreleased/configuration/config-sections/flags-polaris_features.md
@@ -587,6 +587,15 @@ The list of supported authentication types for catalog federation
---
+##### `polaris.features."SUPPORTED_S3_CREDENTIAL_VENDING_MECHANISMS"`
+
+The credential vending mechanisms an S3 catalog in this realm may name explicitly. Names are case-sensitive and must match a mechanism's identifier exactly, for example STS. Apache Polaris supports STS, which is STS AssumeRole. A catalog that leaves credentialVendingMechanism empty uses the server's default mechanism and is always allowed. Realm-level only; catalog properties cannot widen it. Enforced at catalog create and update, and when the server builds the storage integration that v [...]
+
+- **Type:** `List<String>`
+- **Default:** `[STS]`
+
+---
+
##### `polaris.features."TABLE_METADATA_CLEANUP_BATCH_SIZE"`
Metadata batch size for tasks that clean up dropped tables' files.
diff --git a/spec/polaris-management-service.yml b/spec/polaris-management-service.yml
index 484790cb5a..13c5d9ee12 100644
--- a/spec/polaris-management-service.yml
+++ b/spec/polaris-management-service.yml
@@ -1204,6 +1204,18 @@ components:
type: boolean
description: >-
if set to `true`, instructs Polaris Servers to avoid adding KMS key policies.
+ credentialVendingMechanism:
+ type: string
+ example: "STS"
+ description: |
+ How Polaris vends subscoped storage credentials for this catalog. Leave the field empty
+ to use the server's default mechanism, which is STS in Apache Polaris.
+
+ Apache Polaris supports these values:
+ - `STS`: STS AssumeRole against the configured role.
+
+ Later releases may add values. The server returns HTTP 400 for a value it does not
+ support.
AzureStorageConfigInfo:
type: object