SigV4 Auth Support for Catalog Federation - Part 3: Service Identity Info Injection (#2523)

This PR introduces service identity management for SigV4 Auth Support for Catalog Federation. Unlike user-supplied parameters, the service identity represents the identity of the Polaris service itself and should be managed by Polaris.

* Service Identity Injection

* Return injected service identity info in response

* Use AwsCredentialsProvider to retrieve the credentials

* Move some logic to ServiceIdentityConfiguration

* Rename ServiceIdentityRegistry to ServiceIdentityProvider

* Rename ResolvedServiceIdentity to ServiceIdentityCredential

* Simplify the logic and add more tests

* Use SecretReference and fix some small issues

* Disable Catalog Federation

Author: XJDKC

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -303,7 +303,8 @@ public static void enforceFeatureEnabledOrThrow(
               .defaultValue(
                   List.of(
                       AuthenticationParameters.AuthenticationTypeEnum.OAUTH.name(),
-                      AuthenticationParameters.AuthenticationTypeEnum.BEARER.name()))
+                      AuthenticationParameters.AuthenticationTypeEnum.BEARER.name(),
+                      AuthenticationParameters.AuthenticationTypeEnum.SIGV4.name()))
               .buildFeatureConfiguration();
 
   public static final FeatureConfiguration<Integer> ICEBERG_COMMIT_MAX_RETRIES =

polaris-core/src/main/java/org/apache/polaris/core/connection/ConnectionConfigInfoDpo.java

@@ -41,6 +41,7 @@
 import org.apache.polaris.core.connection.iceberg.IcebergCatalogPropertiesProvider;
 import org.apache.polaris.core.connection.iceberg.IcebergRestConnectionConfigInfoDpo;
 import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
+import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
 import org.apache.polaris.core.secrets.SecretReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -225,9 +226,10 @@ public abstract ConnectionConfigInfoDpo withServiceIdentity(
       @Nonnull ServiceIdentityInfoDpo serviceIdentityInfo);
 
   /**
-   * Produces the correponding API-model ConnectionConfigInfo for this persistence object; many
+   * Produces the corresponding API-model ConnectionConfigInfo for this persistence object; many
    * fields are one-to-one direct mappings, but some fields, such as secretReferences, might only be
    * applicable/present in the persistence object, but not the API model object.
    */
-  public abstract ConnectionConfigInfo asConnectionConfigInfoModel();
+  public abstract ConnectionConfigInfo asConnectionConfigInfoModel(
+      ServiceIdentityProvider serviceIdentityProvider);
 }

polaris-core/src/main/java/org/apache/polaris/core/connection/hadoop/HadoopConnectionConfigInfoDpo.java

@@ -32,6 +32,7 @@
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
 import org.apache.polaris.core.connection.ConnectionType;
 import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
+import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
@@ -88,7 +89,8 @@ public ConnectionConfigInfoDpo withServiceIdentity(
   }
 
   @Override
-  public ConnectionConfigInfo asConnectionConfigInfoModel() {
+  public ConnectionConfigInfo asConnectionConfigInfoModel(
+      ServiceIdentityProvider serviceIdentityProvider) {
     return HadoopConnectionConfigInfo.builder()
         .setConnectionType(ConnectionConfigInfo.ConnectionTypeEnum.HADOOP)
         .setUri(getUri())
@@ -97,7 +99,9 @@ public ConnectionConfigInfo asConnectionConfigInfoModel() {
             getAuthenticationParameters().asAuthenticationParametersModel())
         .setServiceIdentity(
             Optional.ofNullable(getServiceIdentity())
-                .map(ServiceIdentityInfoDpo::asServiceIdentityInfoModel)
+                .map(
+                    serviceIdentityInfoDpo ->
+                        serviceIdentityInfoDpo.asServiceIdentityInfoModel(serviceIdentityProvider))
                 .orElse(null))
         .build();
   }

polaris-core/src/main/java/org/apache/polaris/core/connection/hive/HiveConnectionConfigInfoDpo.java

@@ -24,13 +24,15 @@
 import jakarta.annotation.Nullable;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.polaris.core.admin.model.ConnectionConfigInfo;
 import org.apache.polaris.core.admin.model.HiveConnectionConfigInfo;
 import org.apache.polaris.core.connection.AuthenticationParametersDpo;
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
 import org.apache.polaris.core.connection.ConnectionType;
 import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
+import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
@@ -88,13 +90,20 @@ public ConnectionConfigInfoDpo withServiceIdentity(
   }
 
   @Override
-  public ConnectionConfigInfo asConnectionConfigInfoModel() {
+  public ConnectionConfigInfo asConnectionConfigInfoModel(
+      ServiceIdentityProvider serviceIdentityProvider) {
     return HiveConnectionConfigInfo.builder()
         .setConnectionType(ConnectionConfigInfo.ConnectionTypeEnum.HIVE)
         .setUri(getUri())
         .setWarehouse(getWarehouse())
         .setAuthenticationParameters(
             getAuthenticationParameters().asAuthenticationParametersModel())
+        .setServiceIdentity(
+            Optional.ofNullable(getServiceIdentity())
+                .map(
+                    serviceIdentityInfoDpo ->
+                        serviceIdentityInfoDpo.asServiceIdentityInfoModel(serviceIdentityProvider))
+                .orElse(null))
         .build();
   }
 }

polaris-core/src/main/java/org/apache/polaris/core/connection/iceberg/IcebergRestConnectionConfigInfoDpo.java

@@ -32,6 +32,7 @@
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
 import org.apache.polaris.core.connection.ConnectionType;
 import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
+import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
@@ -80,7 +81,8 @@ public ConnectionConfigInfoDpo withServiceIdentity(
   }
 
   @Override
-  public ConnectionConfigInfo asConnectionConfigInfoModel() {
+  public ConnectionConfigInfo asConnectionConfigInfoModel(
+      ServiceIdentityProvider serviceIdentityProvider) {
     return IcebergRestConnectionConfigInfo.builder()
         .setConnectionType(ConnectionConfigInfo.ConnectionTypeEnum.ICEBERG_REST)
         .setUri(getUri())
@@ -89,7 +91,9 @@ public ConnectionConfigInfo asConnectionConfigInfoModel() {
             getAuthenticationParameters().asAuthenticationParametersModel())
         .setServiceIdentity(
             Optional.ofNullable(getServiceIdentity())
-                .map(ServiceIdentityInfoDpo::asServiceIdentityInfoModel)
+                .map(
+                    serviceIdentityInfoDpo ->
+                        serviceIdentityInfoDpo.asServiceIdentityInfoModel(serviceIdentityProvider))
                 .orElse(null))
         .build();
   }

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

@@ -43,6 +43,8 @@
 import org.apache.polaris.core.config.BehaviorChangeConfiguration;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
+import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
 import org.apache.polaris.core.secrets.SecretReference;
 import org.apache.polaris.core.storage.FileStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
@@ -108,6 +110,10 @@ public static CatalogEntity fromCatalog(RealmConfig realmConfig, Catalog catalog
   }
 
   public Catalog asCatalog() {
+    return this.asCatalog(null);
+  }
+
+  public Catalog asCatalog(ServiceIdentityProvider serviceIdentityProvider) {
     Map<String, String> internalProperties = getInternalPropertiesAsMap();
     Catalog.TypeEnum catalogType =
         Optional.ofNullable(internalProperties.get(CATALOG_TYPE_PROPERTY))
@@ -118,6 +124,12 @@ public Catalog asCatalog() {
         CatalogProperties.builder(propertiesMap.get(DEFAULT_BASE_LOCATION_KEY))
             .putAll(propertiesMap)
             .build();
+
+    // Right now, only external catalog may use ServiceIdentityProvider to resolve identity
+    Preconditions.checkState(
+        catalogType != Catalog.TypeEnum.EXTERNAL || serviceIdentityProvider != null,
+        "%s catalog needs ServiceIdentityProvider to resolve service identities",
+        Catalog.TypeEnum.EXTERNAL);
     return catalogType == Catalog.TypeEnum.EXTERNAL
         ? ExternalCatalog.builder()
             .setType(Catalog.TypeEnum.EXTERNAL)
@@ -127,7 +139,7 @@ public Catalog asCatalog() {
             .setLastUpdateTimestamp(getLastUpdateTimestamp())
             .setEntityVersion(getEntityVersion())
             .setStorageConfigInfo(getStorageInfo(internalProperties))
-            .setConnectionConfigInfo(getConnectionInfo(internalProperties))
+            .setConnectionConfigInfo(getConnectionInfo(internalProperties, serviceIdentityProvider))
             .build()
         : PolarisCatalog.builder()
             .setType(Catalog.TypeEnum.INTERNAL)
@@ -187,11 +199,12 @@ private StorageConfigInfo getStorageInfo(Map<String, String> internalProperties)
     return null;
   }
 
-  private ConnectionConfigInfo getConnectionInfo(Map<String, String> internalProperties) {
+  private ConnectionConfigInfo getConnectionInfo(
+      Map<String, String> internalProperties, ServiceIdentityProvider serviceIdentityProvider) {
     if (internalProperties.containsKey(
         PolarisEntityConstants.getConnectionConfigInfoPropertyName())) {
       ConnectionConfigInfoDpo configInfo = getConnectionConfigInfoDpo();
-      return configInfo.asConnectionConfigInfoModel();
+      return configInfo.asConnectionConfigInfoModel(serviceIdentityProvider);
     }
     return null;
   }
@@ -352,11 +365,13 @@ private void validateMaxAllowedLocations(
 
     public Builder setConnectionConfigInfoDpoWithSecrets(
         ConnectionConfigInfo connectionConfigurationModel,
-        Map<String, SecretReference> secretReferences) {
+        Map<String, SecretReference> secretReferences,
+        ServiceIdentityInfoDpo serviceIdentityInfoDpo) {
       if (connectionConfigurationModel != null) {
         ConnectionConfigInfoDpo config =
             ConnectionConfigInfoDpo.fromConnectionConfigInfoModelWithSecrets(
-                connectionConfigurationModel, secretReferences);
+                    connectionConfigurationModel, secretReferences)
+                .withServiceIdentity(serviceIdentityInfoDpo);
         internalProperties.put(
             PolarisEntityConstants.getConnectionConfigInfoPropertyName(), config.serialize());
       }

polaris-core/src/main/java/org/apache/polaris/core/identity/credential/AwsIamServiceIdentityCredential.java

@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.identity.credential;
+
+import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
+import org.apache.polaris.core.admin.model.AwsIamServiceIdentityInfo;
+import org.apache.polaris.core.admin.model.ServiceIdentityInfo;
+import org.apache.polaris.core.identity.ServiceIdentityType;
+import org.apache.polaris.core.identity.dpo.AwsIamServiceIdentityInfoDpo;
+import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
+import org.apache.polaris.core.secrets.SecretReference;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+
+/**
+ * Represents an AWS IAM service identity credential used by Polaris to authenticate to AWS
+ * services.
+ *
+ * <p>This credential encapsulates:
+ *
+ * <ul>
+ *   <li>The IAM ARN (role or user) representing the Polaris service identity
+ *   <li>An {@link AwsCredentialsProvider} that supplies AWS credentials (access key, secret key,
+ *       and optional session token)
+ * </ul>
+ *
+ * <p>Polaris uses this identity to assume customer-provided IAM roles when accessing remote
+ * catalogs with SigV4 authentication. The {@link AwsCredentialsProvider} can be configured to use
+ * either:
+ *
+ * <ul>
+ *   <li>Static credentials (for testing or single-tenant deployments)
+ *   <li>DefaultCredentialsProvider (which chains through various AWS credential sources)
+ *   <li>Custom credential providers (for vendor-specific secret management)
+ * </ul>
+ */
+public class AwsIamServiceIdentityCredential extends ServiceIdentityCredential {
+
+  /** IAM role or user ARN representing the Polaris service identity. */
+  private final String iamArn;
+
+  /** AWS credentials provider for accessing AWS services. */
+  private final AwsCredentialsProvider awsCredentialsProvider;
+
+  public AwsIamServiceIdentityCredential(@Nullable String iamArn) {
+    this(null, iamArn, DefaultCredentialsProvider.builder().build());
+  }
+
+  public AwsIamServiceIdentityCredential(
+      @Nullable String iamArn, @Nonnull AwsCredentialsProvider awsCredentialsProvider) {
+    this(null, iamArn, awsCredentialsProvider);
+  }
+
+  public AwsIamServiceIdentityCredential(
+      @Nullable SecretReference secretReference,
+      @Nullable String iamArn,
+      @Nonnull AwsCredentialsProvider awsCredentialsProvider) {
+    super(ServiceIdentityType.AWS_IAM, secretReference);
+    this.iamArn = iamArn;
+    this.awsCredentialsProvider = awsCredentialsProvider;
+  }
+
+  public @Nullable String getIamArn() {
+    return iamArn;
+  }
+
+  public @Nonnull AwsCredentialsProvider getAwsCredentialsProvider() {
+    return awsCredentialsProvider;
+  }
+
+  @Override
+  public @Nonnull ServiceIdentityInfoDpo asServiceIdentityInfoDpo() {
+    return new AwsIamServiceIdentityInfoDpo(getIdentityInfoReference());
+  }
+
+  @Override
+  public @Nonnull ServiceIdentityInfo asServiceIdentityInfoModel() {
+    return AwsIamServiceIdentityInfo.builder()
+        .setIdentityType(ServiceIdentityInfo.IdentityTypeEnum.AWS_IAM)
+        .setIamArn(getIamArn())
+        .build();
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/identity/credential/ServiceIdentityCredential.java

@@ -0,0 +1,91 @@
+/*
+ * 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.identity.credential;
+
+import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
+import org.apache.polaris.core.admin.model.ServiceIdentityInfo;
+import org.apache.polaris.core.identity.ServiceIdentityType;
+import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
+import org.apache.polaris.core.secrets.SecretReference;
+import software.amazon.awssdk.annotations.NotNull;
+
+/**
+ * Represents a service identity credential used by Polaris to authenticate to external systems.
+ *
+ * <p>This class encapsulates both the service identity metadata (e.g., AWS IAM ARN) and the
+ * associated credentials (e.g., AWS access keys) needed to authenticate as the Polaris service when
+ * accessing external catalog services.
+ *
+ * <p>The credential contains:
+ *
+ * <ul>
+ *   <li>Identity type (e.g., AWS_IAM)
+ *   <li>A {@link SecretReference} that serves as a unique identifier for this service identity
+ *       instance (used for lookups and persistence)
+ *   <li>The actual authentication credentials (implementation-specific, e.g.,
+ *       AwsCredentialsProvider)
+ * </ul>
+ */
+public abstract class ServiceIdentityCredential {
+  private final ServiceIdentityType identityType;
+  private SecretReference identityInfoReference;
+
+  public ServiceIdentityCredential(@Nonnull ServiceIdentityType identityType) {
+    this(identityType, null);
+  }
+
+  public ServiceIdentityCredential(
+      @Nonnull ServiceIdentityType identityType, @Nullable SecretReference identityInfoReference) {
+    this.identityType = identityType;
+    this.identityInfoReference = identityInfoReference;
+  }
+
+  public @NotNull ServiceIdentityType getIdentityType() {
+    return identityType;
+  }
+
+  public @Nonnull SecretReference getIdentityInfoReference() {
+    return identityInfoReference;
+  }
+
+  public void setIdentityInfoReference(@NotNull SecretReference identityInfoReference) {
+    this.identityInfoReference = identityInfoReference;
+  }
+
+  /**
+   * Converts this service identity credential into its corresponding persisted form (DPO).
+   *
+   * <p>The DPO contains only a reference to the credential, not the credential itself, as the
+   * actual secrets are managed externally.
+   *
+   * @return The persistence object representation
+   */
+  public abstract @Nonnull ServiceIdentityInfoDpo asServiceIdentityInfoDpo();
+
+  /**
+   * Converts this service identity credential into its API model representation.
+   *
+   * <p>The model contains identity information (e.g., IAM ARN) but excludes sensitive credentials
+   * such as access keys or session tokens.
+   *
+   * @return The API model representation for client responses
+   */
+  public abstract @Nonnull ServiceIdentityInfo asServiceIdentityInfoModel();
+}