[Splitting] Initial SigV4 Auth Support for Catalog Federation

persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/EclipseLinkPolarisMetaStoreManagerFactory.java

@@ -26,6 +26,7 @@
 import java.nio.file.Path;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.identity.mutation.EntityMutationEngine;
 import org.apache.polaris.core.persistence.LocalPolarisMetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.bootstrap.RootCredentialsSet;
@@ -44,14 +45,16 @@ public class EclipseLinkPolarisMetaStoreManagerFactory
 
   @Inject EclipseLinkConfiguration eclipseLinkConfiguration;
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
+  @Inject EntityMutationEngine entityMutationEngine;
 
   protected EclipseLinkPolarisMetaStoreManagerFactory() {
-    this(null);
+    this(null, null);
   }
 
   @Inject
-  protected EclipseLinkPolarisMetaStoreManagerFactory(PolarisDiagnostics diagnostics) {
-    super(diagnostics);
+  protected EclipseLinkPolarisMetaStoreManagerFactory(
+      PolarisDiagnostics diagnostics, EntityMutationEngine entityMutationEngine) {
+    super(diagnostics, entityMutationEngine);
   }
 
   @Override

persistence/eclipselink/src/test/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkMetaStoreManagerTest.java

@@ -39,6 +39,7 @@
 import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
+import org.apache.polaris.core.identity.mutation.NoOpEntityMutationEngine;
 import org.apache.polaris.core.persistence.BasePolarisMetaStoreManagerTest;
 import org.apache.polaris.core.persistence.PolarisTestMetaStoreManager;
 import org.apache.polaris.core.persistence.transactional.TransactionalMetaStoreManagerImpl;
@@ -96,6 +97,7 @@ protected PolarisTestMetaStoreManager createPolarisTestMetaStoreManager() {
             session,
             diagServices,
             new PolarisConfigurationStore() {},
+            new NoOpEntityMutationEngine(),
             timeSource.withZone(ZoneId.systemDefault())));
   }
 

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java

@@ -38,6 +38,7 @@
 import org.apache.polaris.core.entity.PolarisEntityConstants;
 import org.apache.polaris.core.entity.PolarisEntitySubType;
 import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.identity.mutation.EntityMutationEngine;
 import org.apache.polaris.core.persistence.AtomicOperationMetaStoreManager;
 import org.apache.polaris.core.persistence.BasePersistence;
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
@@ -74,6 +75,7 @@ public class JdbcMetaStoreManagerFactory implements MetaStoreManagerFactory {
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
   @Inject Instance<DataSource> dataSource;
   @Inject RelationalJdbcConfiguration relationalJdbcConfiguration;
+  @Inject EntityMutationEngine entityMutationEngine;
 
   protected JdbcMetaStoreManagerFactory() {}
 
@@ -158,7 +160,8 @@ public Map<String, BaseResult> purgeRealms(Iterable<String> realms) {
       PolarisMetaStoreManager metaStoreManager = getOrCreateMetaStoreManager(realmContext);
       BasePersistence session = getOrCreateSessionSupplier(realmContext).get();
 
-      PolarisCallContext callContext = new PolarisCallContext(realmContext, session, diagServices);
+      PolarisCallContext callContext =
+          new PolarisCallContext(realmContext, session, diagServices, entityMutationEngine);
       BaseResult result = metaStoreManager.purge(callContext);
       results.put(realm, result);
 
@@ -229,7 +232,8 @@ private PrincipalSecretsResult bootstrapServiceAndCreatePolarisPrincipalForRealm
         new PolarisCallContext(
             realmContext,
             sessionSupplierMap.get(realmContext.getRealmIdentifier()).get(),
-            diagServices);
+            diagServices,
+            entityMutationEngine);
     if (CallContext.getCurrentContext() == null) {
       CallContext.setCurrentContext(polarisContext);
     }
@@ -280,7 +284,8 @@ private void checkPolarisServiceBootstrappedForRealm(
         new PolarisCallContext(
             realmContext,
             sessionSupplierMap.get(realmContext.getRealmIdentifier()).get(),
-            diagServices);
+            diagServices,
+            entityMutationEngine);
     if (CallContext.getCurrentContext() == null) {
       CallContext.setCurrentContext(polarisContext);
     }

persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/AtomicMetastoreManagerWithJdbcBasePersistenceImplTest.java

@@ -29,6 +29,7 @@
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.identity.mutation.NoOpEntityMutationEngine;
 import org.apache.polaris.core.persistence.AtomicOperationMetaStoreManager;
 import org.apache.polaris.core.persistence.BasePolarisMetaStoreManagerTest;
 import org.apache.polaris.core.persistence.PolarisTestMetaStoreManager;
@@ -71,6 +72,7 @@ protected PolarisTestMetaStoreManager createPolarisTestMetaStoreManager() {
             basePersistence,
             diagServices,
             new PolarisConfigurationStore() {},
+            new NoOpEntityMutationEngine(),
             timeSource.withZone(ZoneId.systemDefault())));
   }
 

polaris-core/src/main/java/org/apache/polaris/core/PolarisCallContext.java

@@ -24,6 +24,7 @@
 import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.identity.mutation.EntityMutationEngine;
 import org.apache.polaris.core.persistence.BasePersistence;
 
 /**
@@ -40,6 +41,8 @@ public class PolarisCallContext implements CallContext {
 
   private final PolarisConfigurationStore configurationStore;
 
+  private final EntityMutationEngine entityMutationEngine;
+
   private final Clock clock;
 
   // will make it final once we remove deprecated constructor
@@ -50,22 +53,26 @@ public PolarisCallContext(
       @Nonnull BasePersistence metaStore,
       @Nonnull PolarisDiagnostics diagServices,
       @Nonnull PolarisConfigurationStore configurationStore,
+      @Nonnull EntityMutationEngine entityMutationEngine,
       @Nonnull Clock clock) {
     this.realmContext = realmContext;
     this.metaStore = metaStore;
     this.diagServices = diagServices;
     this.configurationStore = configurationStore;
+    this.entityMutationEngine = entityMutationEngine;
     this.clock = clock;
   }
 
   public PolarisCallContext(
       @Nonnull RealmContext realmContext,
       @Nonnull BasePersistence metaStore,
-      @Nonnull PolarisDiagnostics diagServices) {
+      @Nonnull PolarisDiagnostics diagServices,
+      @Nonnull EntityMutationEngine entityMutationEngine) {
     this.realmContext = realmContext;
     this.metaStore = metaStore;
     this.diagServices = diagServices;
     this.configurationStore = new PolarisConfigurationStore() {};
+    this.entityMutationEngine = entityMutationEngine;
     this.clock = Clock.system(ZoneId.systemDefault());
   }
 
@@ -81,6 +88,10 @@ public PolarisConfigurationStore getConfigurationStore() {
     return configurationStore;
   }
 
+  public EntityMutationEngine getEntityMutationEngine() {
+    return entityMutationEngine;
+  }
+
   public Clock getClock() {
     return clock;
   }
@@ -105,6 +116,11 @@ public PolarisCallContext copy() {
     String realmId = this.realmContext.getRealmIdentifier();
     RealmContext realmContext = () -> realmId;
     return new PolarisCallContext(
-        realmContext, this.metaStore, this.diagServices, this.configurationStore, this.clock);
+        realmContext,
+        this.metaStore,
+        this.diagServices,
+        this.configurationStore,
+        this.entityMutationEngine,
+        this.clock);
   }
 }

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

@@ -18,13 +18,16 @@
  */
 package org.apache.polaris.core.connection;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import jakarta.annotation.Nonnull;
 import java.util.Map;
 import org.apache.polaris.core.admin.model.AuthenticationParameters;
 import org.apache.polaris.core.admin.model.BearerAuthenticationParameters;
 import org.apache.polaris.core.admin.model.OAuthClientCredentialsParameters;
+import org.apache.polaris.core.admin.model.SigV4AuthenticationParameters;
 import org.apache.polaris.core.connection.iceberg.IcebergCatalogPropertiesProvider;
 import org.apache.polaris.core.secrets.UserSecretReference;
 
@@ -39,6 +42,7 @@
 @JsonSubTypes({
   @JsonSubTypes.Type(value = OAuthClientCredentialsParametersDpo.class, name = "1"),
   @JsonSubTypes.Type(value = BearerAuthenticationParametersDpo.class, name = "2"),
+  @JsonSubTypes.Type(value = SigV4AuthenticationParametersDpo.class, name = "3"),
 })
 public abstract class AuthenticationParametersDpo implements IcebergCatalogPropertiesProvider {
 
@@ -57,7 +61,12 @@ public int getAuthenticationTypeCode() {
     return authenticationTypeCode;
   }
 
-  public abstract AuthenticationParameters asAuthenticationParametersModel();
+  @JsonIgnore
+  public AuthenticationType getAuthenticationType() {
+    return AuthenticationType.fromCode(authenticationTypeCode);
+  }
+
+  public abstract @Nonnull AuthenticationParameters asAuthenticationParametersModel();
 
   public static AuthenticationParametersDpo fromAuthenticationParametersModelWithSecrets(
       AuthenticationParameters authenticationParameters,
@@ -81,6 +90,18 @@ public static AuthenticationParametersDpo fromAuthenticationParametersModelWithS
             new BearerAuthenticationParametersDpo(
                 secretReferences.get(INLINE_BEARER_TOKEN_REFERENCE_KEY));
         break;
+      case SIGV4:
+        // SigV4 authentication is not secret-based
+        SigV4AuthenticationParameters sigV4AuthenticationParametersModel =
+            (SigV4AuthenticationParameters) authenticationParameters;
+        config =
+            new SigV4AuthenticationParametersDpo(
+                sigV4AuthenticationParametersModel.getRoleArn(),
+                sigV4AuthenticationParametersModel.getRoleSessionName(),
+                sigV4AuthenticationParametersModel.getExternalId(),
+                sigV4AuthenticationParametersModel.getSigningRegion(),
+                sigV4AuthenticationParametersModel.getSigningName());
+        break;
       default:
         throw new IllegalStateException(
             "Unsupported authentication type: " + authenticationParameters.getAuthenticationType());

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

@@ -33,6 +33,7 @@ public enum AuthenticationType {
   NULL_TYPE(0),
   OAUTH(1),
   BEARER(2),
+  SIGV4(3),
   ;
 
   private static final AuthenticationType[] REVERSE_MAPPING_ARRAY;
@@ -65,7 +66,7 @@ public enum AuthenticationType {
    * NULL_TYPE if not found
    *
    * @param authTypeCode code associated to the authentication type
-   * @return ConnectionType corresponding to that code or null if mapping not found
+   * @return AuthenticationType corresponding to that code or null if mapping not found
    */
   public static @Nonnull AuthenticationType fromCode(int authTypeCode) {
     // ensure it is within bounds

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.UserSecretReference;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
@@ -50,13 +51,13 @@ 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);
   }
 
   @Override
-  public AuthenticationParameters asAuthenticationParametersModel() {
+  public @Nonnull AuthenticationParameters asAuthenticationParametersModel() {
     return BearerAuthenticationParameters.builder()
         .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.BEARER)
         .build();

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

@@ -18,6 +18,7 @@
  */
 package org.apache.polaris.core.connection;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
@@ -38,6 +39,7 @@
 import org.apache.polaris.core.connection.hadoop.HadoopConnectionConfigInfoDpo;
 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.secrets.UserSecretReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -66,22 +68,29 @@ public abstract class ConnectionConfigInfoDpo implements IcebergCatalogPropertie
   // The authentication parameters for the connection
   private final AuthenticationParametersDpo authenticationParameters;
 
+  // The Polaris service identity info of the connection
+  private final ServiceIdentityInfoDpo serviceIdentity;
+
   public ConnectionConfigInfoDpo(
       @JsonProperty(value = "connectionTypeCode", required = true) int connectionTypeCode,
       @JsonProperty(value = "uri", required = true) @Nonnull String uri,
       @JsonProperty(value = "authenticationParameters", required = true) @Nonnull
-          AuthenticationParametersDpo authenticationParameters) {
-    this(connectionTypeCode, uri, authenticationParameters, true);
+          AuthenticationParametersDpo authenticationParameters,
+      @JsonProperty(value = "serviceIdentity", required = false) @Nullable
+          ServiceIdentityInfoDpo serviceIdentity) {
+    this(connectionTypeCode, uri, authenticationParameters, serviceIdentity, true);
   }
 
   protected ConnectionConfigInfoDpo(
       int connectionTypeCode,
       @Nonnull String uri,
       @Nonnull AuthenticationParametersDpo authenticationParameters,
+      @Nullable ServiceIdentityInfoDpo serviceIdentity,
       boolean validateUri) {
     this.connectionTypeCode = connectionTypeCode;
     this.uri = uri;
     this.authenticationParameters = authenticationParameters;
+    this.serviceIdentity = serviceIdentity;
     if (validateUri) {
       validateUri(uri);
     }
@@ -91,6 +100,11 @@ public int getConnectionTypeCode() {
     return connectionTypeCode;
   }
 
+  @JsonIgnore
+  public ConnectionType getConnectionType() {
+    return ConnectionType.fromCode(connectionTypeCode);
+  }
+
   public String getUri() {
     return uri;
   }
@@ -99,6 +113,10 @@ public AuthenticationParametersDpo getAuthenticationParameters() {
     return authenticationParameters;
   }
 
+  public @Nullable ServiceIdentityInfoDpo getServiceIdentity() {
+    return serviceIdentity;
+  }
+
   private static final ObjectMapper DEFAULT_MAPPER;
 
   static {
@@ -157,6 +175,7 @@ public static ConnectionConfigInfoDpo fromConnectionConfigInfoModelWithSecrets(
             new IcebergRestConnectionConfigInfoDpo(
                 icebergRestConfigModel.getUri(),
                 authenticationParameters,
+                null /*Service Identity Info*/,
                 icebergRestConfigModel.getRemoteCatalogName());
         break;
       case HADOOP:
@@ -169,6 +188,7 @@ public static ConnectionConfigInfoDpo fromConnectionConfigInfoModelWithSecrets(
             new HadoopConnectionConfigInfoDpo(
                 hadoopConfigModel.getUri(),
                 authenticationParameters,
+                null /*Service Identity Info*/,
                 hadoopConfigModel.getWarehouse());
         break;
       default:
@@ -178,6 +198,9 @@ public static ConnectionConfigInfoDpo fromConnectionConfigInfoModelWithSecrets(
     return config;
   }
 
+  public abstract ConnectionConfigInfoDpo withServiceIdentity(
+      @Nonnull ServiceIdentityInfoDpo serviceIdentityInfo);
+
   /**
    * Produces the correponding API-model ConnectionConfigInfo for this persistence object; many
    * fields are one-to-one direct mappings, but some fields, such as secretReferences, might only be