SigV4 Auth Support for Catalog Federation - Part 2: Connection Config Persistence

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

@@ -56,7 +56,7 @@ public BearerAuthenticationParametersDpo(
   }
 
   @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,15 @@ public static ConnectionConfigInfoDpo fromConnectionConfigInfoModelWithSecrets(
     return config;
   }
 
+  /**
+   * Creates a new copy of the ConnectionConfigInfoDpo with the given service identity info.
+   *
+   * @param serviceIdentityInfo The service identity info to set.
+   * @return A new copy of the ConnectionConfigInfoDpo with the given service identity info.
+   */
+  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

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

@@ -86,7 +86,7 @@ public OAuthClientCredentialsParametersDpo(
     return clientSecretReference;
   }
 
-  public @Nonnull List<String> getScopes() {
+  public @Nullable List<String> getScopes() {
     return scopes;
   }
 
@@ -115,7 +115,7 @@ public OAuthClientCredentialsParametersDpo(
   }
 
   @Override
-  public AuthenticationParameters asAuthenticationParametersModel() {
+  public @Nonnull AuthenticationParameters asAuthenticationParametersModel() {
     return OAuthClientCredentialsParameters.builder()
         .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.OAUTH)
         .setTokenUri(getTokenUri())

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

@@ -0,0 +1,131 @@
+/*
+ * 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.connection;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableMap;
+import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
+import java.util.Map;
+import org.apache.iceberg.aws.AwsProperties;
+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.secrets.UserSecretsManager;
+
+/**
+ * The internal persistence-object counterpart to SigV4AuthenticationParameters defined in the API
+ * model.
+ */
+public class SigV4AuthenticationParametersDpo extends AuthenticationParametersDpo {
+
+  // The aws IAM role arn assumed by polaris userArn when signing requests
+  @JsonProperty(value = "roleArn")
+  private final String roleArn;
+
+  // The session name used when assuming the role
+  @JsonProperty(value = "roleSessionName")
+  private final String roleSessionName;
+
+  // An optional external id used to establish a trust relationship with AWS in the trust policy
+  @JsonProperty(value = "externalId")
+  private final String externalId;
+
+  // Region to be used by the SigV4 protocol for signing requests
+  @JsonProperty(value = "signingRegion")
+  private final String signingRegion;
+
+  // The service name to be used by the SigV4 protocol for signing requests, the default signing
+  // name is "execute-api" is if not provided
+  @JsonProperty(value = "signingName")
+  private final String signingName;
+
+  public SigV4AuthenticationParametersDpo(
+      @JsonProperty(value = "roleArn", required = true) String roleArn,
+      @JsonProperty(value = "roleSessionName", required = false) String roleSessionName,
+      @JsonProperty(value = "externalId", required = false) String externalId,
+      @JsonProperty(value = "signingRegion", required = true) String signingRegion,
+      @JsonProperty(value = "signingName", required = false) String signingName) {
+    super(AuthenticationType.SIGV4.getCode());
+    this.roleArn = roleArn;
+    this.roleSessionName = roleSessionName;
+    this.externalId = externalId;
+    this.signingRegion = signingRegion;
+    this.signingName = signingName;
+  }
+
+  public @Nonnull String getRoleArn() {
+    return roleArn;
+  }
+
+  public @Nullable String getRoleSessionName() {
+    return roleSessionName;
+  }
+
+  public @Nullable String getExternalId() {
+    return externalId;
+  }
+
+  public @Nonnull String getSigningRegion() {
+    return signingRegion;
+  }
+
+  public @Nullable String getSigningName() {
+    return signingName;
+  }
+
+  @Nonnull
+  @Override
+  public Map<String, String> asIcebergCatalogProperties(UserSecretsManager secretsManager) {
+    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
+    return builder.build();
+  }
+
+  @Override
+  public @Nonnull AuthenticationParameters asAuthenticationParametersModel() {
+    return SigV4AuthenticationParameters.builder()
+        .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.SIGV4)
+        .setRoleArn(getRoleArn())
+        .setRoleSessionName(getRoleSessionName())
+        .setExternalId(getExternalId())
+        .setSigningRegion(getSigningRegion())
+        .setSigningName(getSigningName())
+        .build();
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("authenticationTypeCode", getAuthenticationTypeCode())
+        .add("roleArn", getRoleArn())
+        .add("roleSessionName", getRoleSessionName())
+        .add("externalId", getExternalId())
+        .add("signingRegion", getSigningRegion())
+        .add("signingName", getSigningName())
+        .toString();
+  }
+}

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

@@ -24,12 +24,14 @@
 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.HadoopConnectionConfigInfo;
 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.secrets.UserSecretsManager;
 
 /**
@@ -44,8 +46,10 @@ public HadoopConnectionConfigInfoDpo(
       @JsonProperty(value = "uri", required = true) @Nonnull String uri,
       @JsonProperty(value = "authenticationParameters", required = true) @Nonnull
           AuthenticationParametersDpo authenticationParameters,
+      @JsonProperty(value = "serviceIdentity", required = false) @Nullable
+          ServiceIdentityInfoDpo serviceIdentityInfo,
       @JsonProperty(value = "warehouse", required = false) @Nullable String remoteCatalogName) {
-    super(ConnectionType.HADOOP.getCode(), uri, authenticationParameters);
+    super(ConnectionType.HADOOP.getCode(), uri, authenticationParameters, serviceIdentityInfo);
     this.warehouse = remoteCatalogName;
   }
 
@@ -60,6 +64,7 @@ public String toString() {
         .add("uri", getUri())
         .add("warehouse", getWarehouse())
         .add("authenticationParameters", getAuthenticationParameters().toString())
+        .add("serviceIdentity", getServiceIdentity())
         .toString();
   }
 
@@ -75,6 +80,13 @@ public String toString() {
     return properties;
   }
 
+  @Override
+  public ConnectionConfigInfoDpo withServiceIdentity(
+      @Nonnull ServiceIdentityInfoDpo serviceIdentityInfo) {
+    return new HadoopConnectionConfigInfoDpo(
+        getUri(), getAuthenticationParameters(), serviceIdentityInfo, getWarehouse());
+  }
+
   @Override
   public ConnectionConfigInfo asConnectionConfigInfoModel() {
     return HadoopConnectionConfigInfo.builder()
@@ -83,6 +95,10 @@ public ConnectionConfigInfo asConnectionConfigInfoModel() {
         .setWarehouse(getWarehouse())
         .setAuthenticationParameters(
             getAuthenticationParameters().asAuthenticationParametersModel())
+        .setServiceIdentity(
+            Optional.ofNullable(getServiceIdentity())
+                .map(ServiceIdentityInfoDpo::asServiceIdentityInfoModel)
+                .orElse(null))
         .build();
   }
 }