SigV4 Auth Support for Catalog Federation - Part 4: Connection Credential Manager

extensions/federation/hadoop/src/main/java/org/apache/polaris/extensions/federation/hadoop/HadoopFederatedCatalogFactory.java

@@ -30,6 +30,7 @@
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
 import org.apache.polaris.core.connection.ConnectionType;
 import org.apache.polaris.core.connection.hadoop.HadoopConnectionConfigInfoDpo;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -42,7 +43,9 @@ public class HadoopFederatedCatalogFactory implements ExternalCatalogFactory {
 
   @Override
   public Catalog createCatalog(
-      ConnectionConfigInfoDpo connectionConfigInfoDpo, UserSecretsManager userSecretsManager) {
+      ConnectionConfigInfoDpo connectionConfigInfoDpo,
+      UserSecretsManager userSecretsManager,
+      PolarisCredentialManager polarisCredentialManager) {
     // Currently, Polaris supports Hadoop federation only via IMPLICIT authentication.
     // Hence, prior to initializing the configuration, ensure that the catalog uses
     // IMPLICIT authentication.
@@ -56,7 +59,9 @@ public Catalog createCatalog(
     String warehouse = ((HadoopConnectionConfigInfoDpo) connectionConfigInfoDpo).getWarehouse();
     HadoopCatalog hadoopCatalog = new HadoopCatalog(conf, warehouse);
     hadoopCatalog.initialize(
-        warehouse, connectionConfigInfoDpo.asIcebergCatalogProperties(userSecretsManager));
+        warehouse,
+        connectionConfigInfoDpo.asIcebergCatalogProperties(
+            userSecretsManager, polarisCredentialManager));
     return hadoopCatalog;
   }
 

extensions/federation/hive/src/main/java/org/apache/polaris/extensions/federation/hive/HiveFederatedCatalogFactory.java

@@ -29,6 +29,7 @@
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
 import org.apache.polaris.core.connection.ConnectionType;
 import org.apache.polaris.core.connection.hive.HiveConnectionConfigInfoDpo;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,7 +42,9 @@ public class HiveFederatedCatalogFactory implements ExternalCatalogFactory {
 
   @Override
   public Catalog createCatalog(
-      ConnectionConfigInfoDpo connectionConfigInfoDpo, UserSecretsManager userSecretsManager) {
+      ConnectionConfigInfoDpo connectionConfigInfoDpo,
+      UserSecretsManager userSecretsManager,
+      PolarisCredentialManager polarisCredentialManager) {
     // Currently, Polaris supports Hive federation only via IMPLICIT authentication.
     // Hence, prior to initializing the configuration, ensure that the catalog uses
     // IMPLICIT authentication.
@@ -69,7 +72,9 @@ public Catalog createCatalog(
     // Kerberos instances are not suitable because Kerberos ties a single identity to the server.
     HiveCatalog hiveCatalog = new HiveCatalog();
     hiveCatalog.initialize(
-        warehouse, connectionConfigInfoDpo.asIcebergCatalogProperties(userSecretsManager));
+        warehouse,
+        connectionConfigInfoDpo.asIcebergCatalogProperties(
+            userSecretsManager, polarisCredentialManager));
     return hiveCatalog;
   }
 

polaris-core/src/main/java/org/apache/polaris/core/catalog/ExternalCatalogFactory.java

@@ -20,6 +20,7 @@
 
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
@@ -34,12 +35,16 @@ public interface ExternalCatalogFactory {
    * Creates a catalog handle for the given connection configuration.
    *
    * @param connectionConfig the connection configuration
-   * @param userSecretsManager the user secrets manager for handling credentials
+   * @param userSecretsManager the user secrets manager for handling user-provided credentials
+   * @param polarisCredentialManager the credential manager for generating temporary credentials
+   *     that Polaris uses to access external systems
    * @return the initialized catalog
    * @throws IllegalStateException if the connection configuration is invalid
    */
   Catalog createCatalog(
-      ConnectionConfigInfoDpo connectionConfig, UserSecretsManager userSecretsManager);
+      ConnectionConfigInfoDpo connectionConfig,
+      UserSecretsManager userSecretsManager,
+      PolarisCredentialManager polarisCredentialManager);
 
   /**
    * Creates a generic table catalog for the given connection configuration.

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

@@ -25,6 +25,7 @@
 import org.apache.iceberg.rest.auth.OAuth2Properties;
 import org.apache.polaris.core.admin.model.AuthenticationParameters;
 import org.apache.polaris.core.admin.model.BearerAuthenticationParameters;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
 import org.apache.polaris.core.secrets.SecretReference;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
@@ -50,7 +51,7 @@ public BearerAuthenticationParametersDpo(
 
   @Override
   public @Nonnull Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager) {
+      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
     String bearerToken = secretsManager.readSecret(getBearerTokenReference());
     return Map.of(OAuth2Properties.TOKEN, bearerToken);
   }

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

@@ -19,9 +19,11 @@
 package org.apache.polaris.core.connection;
 
 import com.google.common.base.MoreObjects;
+import jakarta.annotation.Nonnull;
 import java.util.Map;
 import org.apache.polaris.core.admin.model.AuthenticationParameters;
 import org.apache.polaris.core.admin.model.ImplicitAuthenticationParameters;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
@@ -35,12 +37,13 @@ public ImplicitAuthenticationParametersDpo() {
   }
 
   @Override
-  public Map<String, String> asIcebergCatalogProperties(UserSecretsManager secretsManager) {
+  public @Nonnull Map<String, String> asIcebergCatalogProperties(
+      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
     return Map.of();
   }
 
   @Override
-  public AuthenticationParameters asAuthenticationParametersModel() {
+  public @Nonnull AuthenticationParameters asAuthenticationParametersModel() {
     return ImplicitAuthenticationParameters.builder()
         .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.IMPLICIT)
         .build();

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

@@ -35,6 +35,7 @@
 import org.apache.iceberg.rest.auth.OAuth2Util;
 import org.apache.polaris.core.admin.model.AuthenticationParameters;
 import org.apache.polaris.core.admin.model.OAuthClientCredentialsParameters;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
 import org.apache.polaris.core.secrets.SecretReference;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
@@ -104,7 +105,7 @@ public OAuthClientCredentialsParametersDpo(
 
   @Override
   public @Nonnull Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager) {
+      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
     HashMap<String, String> properties = new HashMap<>();
     if (getTokenUri() != null) {
       properties.put(OAuth2Properties.OAUTH2_SERVER_URI, getTokenUri());

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

@@ -28,6 +28,7 @@
 import org.apache.iceberg.rest.auth.AuthProperties;
 import org.apache.polaris.core.admin.model.AuthenticationParameters;
 import org.apache.polaris.core.admin.model.SigV4AuthenticationParameters;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
@@ -93,15 +94,15 @@ public SigV4AuthenticationParametersDpo(
 
   @Nonnull
   @Override
-  public Map<String, String> asIcebergCatalogProperties(UserSecretsManager secretsManager) {
+  public Map<String, String> asIcebergCatalogProperties(
+      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
     ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
     builder.put(AuthProperties.AUTH_TYPE, AuthProperties.AUTH_TYPE_SIGV4);
     builder.put(AwsProperties.REST_SIGNER_REGION, getSigningRegion());
     if (getSigningName() != null) {
       builder.put(AwsProperties.REST_SIGNING_NAME, getSigningName());
     }
-
-    // TODO: Add a credential manager to assume the role and get the aws session credentials
+    // Connection credentials are handled by ConnectionConfigInfoDpo
     return builder.build();
   }
 

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

@@ -31,6 +31,8 @@
 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.credentials.PolarisCredentialManager;
+import org.apache.polaris.core.credentials.connection.ConnectionCredentials;
 import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
 import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
 import org.apache.polaris.core.secrets.UserSecretsManager;
@@ -71,13 +73,19 @@ public String toString() {
 
   @Override
   public @Nonnull Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager) {
+      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
     HashMap<String, String> properties = new HashMap<>();
     properties.put(CatalogProperties.URI, getUri());
     if (getWarehouse() != null) {
       properties.put(CatalogProperties.WAREHOUSE_LOCATION, getWarehouse());
     }
-    properties.putAll(getAuthenticationParameters().asIcebergCatalogProperties(secretsManager));
+    // Add authentication-specific properties
+    properties.putAll(
+        getAuthenticationParameters()
+            .asIcebergCatalogProperties(secretsManager, credentialManager));
+    // Add connection credentials from Polaris credential manager
+    ConnectionCredentials connectionCredentials = credentialManager.getConnectionCredentials(this);
+    properties.putAll(connectionCredentials.credentials());
     return properties;
   }
 

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

@@ -31,6 +31,8 @@
 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.credentials.PolarisCredentialManager;
+import org.apache.polaris.core.credentials.connection.ConnectionCredentials;
 import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
 import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
 import org.apache.polaris.core.secrets.UserSecretsManager;
@@ -70,14 +72,21 @@ public String toString() {
 
   @Override
   public @Nonnull Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager) {
+      UserSecretsManager secretsManager, PolarisCredentialManager polarisCredentialManager) {
     HashMap<String, String> properties = new HashMap<>();
     properties.put(CatalogProperties.URI, getUri());
     if (getWarehouse() != null) {
       properties.put(CatalogProperties.WAREHOUSE_LOCATION, getWarehouse());
     }
     if (getAuthenticationParameters() != null) {
-      properties.putAll(getAuthenticationParameters().asIcebergCatalogProperties(secretsManager));
+      // Add authentication-specific properties
+      properties.putAll(
+          getAuthenticationParameters()
+              .asIcebergCatalogProperties(secretsManager, polarisCredentialManager));
+      // Add connection credentials from Polaris credential manager
+      ConnectionCredentials connectionCredentials =
+          polarisCredentialManager.getConnectionCredentials(this);
+      properties.putAll(connectionCredentials.credentials());
     }
     return properties;
   }

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

@@ -20,6 +20,7 @@
 
 import jakarta.annotation.Nonnull;
 import java.util.Map;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
@@ -30,5 +31,6 @@
  */
 public interface IcebergCatalogPropertiesProvider {
   @Nonnull
-  Map<String, String> asIcebergCatalogProperties(UserSecretsManager secretsManager);
+  Map<String, String> asIcebergCatalogProperties(
+      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager);
 }

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

@@ -31,6 +31,8 @@
 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.credentials.PolarisCredentialManager;
+import org.apache.polaris.core.credentials.connection.ConnectionCredentials;
 import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
 import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
 import org.apache.polaris.core.secrets.UserSecretsManager;
@@ -63,13 +65,19 @@ public String getRemoteCatalogName() {
 
   @Override
   public @Nonnull Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager) {
+      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
     HashMap<String, String> properties = new HashMap<>();
     properties.put(CatalogProperties.URI, getUri());
     if (getRemoteCatalogName() != null) {
       properties.put(CatalogProperties.WAREHOUSE_LOCATION, getRemoteCatalogName());
     }
-    properties.putAll(getAuthenticationParameters().asIcebergCatalogProperties(secretsManager));
+    // Add authentication-specific properties
+    properties.putAll(
+        getAuthenticationParameters()
+            .asIcebergCatalogProperties(secretsManager, credentialManager));
+    // Add connection credentials from Polaris credential manager
+    ConnectionCredentials connectionCredentials = credentialManager.getConnectionCredentials(this);
+    properties.putAll(connectionCredentials.credentials());
     return properties;
   }
 

polaris-core/src/main/java/org/apache/polaris/core/credentials/PolarisCredentialManager.java

@@ -0,0 +1,52 @@
+/*
+ * 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.credentials;
+
+import org.apache.polaris.core.connection.AuthenticationParametersDpo;
+import org.apache.polaris.core.credentials.connection.ConnectionCredentialVendor;
+import org.apache.polaris.core.identity.credential.ServiceIdentityCredential;
+import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
+
+/**
+ * PolarisCredentialManager is responsible for retrieving the credentials Polaris needs to access
+ * remote services such as federated catalogs and cloud storage.
+ *
+ * <p>It delegates to {@link ConnectionCredentialVendor} implementations that combine
+ * service-managed identity information (e.g., an IAM user Polaris uses) with user-defined
+ * authentication parameters (e.g., roleArn) to generate the credentials required for authentication
+ * with external systems.
+ *
+ * <p>Typical flow for connection credentials:
+ *
+ * <ol>
+ *   <li>The manager selects the appropriate {@link ConnectionCredentialVendor} based on the
+ *       authentication type from the {@link AuthenticationParametersDpo}.
+ *   <li>The vendor resolves the service identity using {@link ServiceIdentityProvider} to obtain a
+ *       {@link ServiceIdentityCredential}.
+ *   <li>The vendor uses the service identity credential together with user-provided authentication
+ *       parameters to obtain temporary access credentials (e.g., via AWS STS AssumeRole).
+ * </ol>
+ *
+ * <p>This design supports both SaaS and self-managed deployments, ensuring a clear separation
+ * between user-provided configuration and Polaris-managed identity. In the future, this interface
+ * will be extended to also manage storage credentials, providing a unified interface for all
+ * credential management needs.
+ */
+public interface PolarisCredentialManager extends ConnectionCredentialVendor {}