Support IMPLICIT authentication type for federated catalogs (#1925)

Previously, the ConnectionConfigInfo required explicit AuthenticationParameters for every federated catalog. However, certain catalogs types that Polaris federates to (either now or in the future) allow `IMPLICIT` authentication, wherein the authentication parameters are picked from the environment or configuration files. This change enables federating to such catalogs without passing dummy secrets. 

The `IMPLICIT` option is guarded by the `SUPPORTED_EXTERNAL_CATALOG_AUTHENTICATION_TYPES`. Hence users may create federated catalogs with `IMPLICIT` authentication only when the administrator explicitly enables this feature.

Author: poojanilangekar

CHANGELOG.md

@@ -37,6 +37,10 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 
 - Added Catalog configuration for S3 and STS endpoints. This also allows using non-AWS S3 implementations.
 
+- The `IMPLICIT` authentication type enables users to create federated catalogs without explicitly
+providing authentication parameters to Polaris. When the authentication type is set to `IMPLICIT`, 
+the authentication parameters are picked from the environment or configuration files. 
+
 ### Changes
 
 ### Deprecations

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

@@ -39,6 +39,7 @@
 @JsonSubTypes({
   @JsonSubTypes.Type(value = OAuthClientCredentialsParametersDpo.class, name = "1"),
   @JsonSubTypes.Type(value = BearerAuthenticationParametersDpo.class, name = "2"),
+  @JsonSubTypes.Type(value = ImplicitAuthenticationParametersDpo.class, name = "3"),
 })
 public abstract class AuthenticationParametersDpo implements IcebergCatalogPropertiesProvider {
 
@@ -81,6 +82,9 @@ public static AuthenticationParametersDpo fromAuthenticationParametersModelWithS
             new BearerAuthenticationParametersDpo(
                 secretReferences.get(INLINE_BEARER_TOKEN_REFERENCE_KEY));
         break;
+      case IMPLICIT:
+        config = new ImplicitAuthenticationParametersDpo();
+        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),
+  IMPLICIT(3),
   ;
 
   private static final AuthenticationType[] REVERSE_MAPPING_ARRAY;

polaris-core/src/main/java/org/apache/polaris/core/connection/ImplicitAuthenticationParametersDpo.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.core.connection;
+
+import com.google.common.base.MoreObjects;
+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.secrets.UserSecretsManager;
+
+/**
+ * The internal persistence-object counterpart to ImplicitAuthenticationParameters defined in the
+ * API model.
+ */
+public class ImplicitAuthenticationParametersDpo extends AuthenticationParametersDpo {
+
+  public ImplicitAuthenticationParametersDpo() {
+    super(AuthenticationType.IMPLICIT.getCode());
+  }
+
+  @Override
+  public Map<String, String> asIcebergCatalogProperties(UserSecretsManager secretsManager) {
+    return Map.of();
+  }
+
+  @Override
+  public AuthenticationParameters asAuthenticationParametersModel() {
+    return ImplicitAuthenticationParameters.builder()
+        .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.IMPLICIT)
+        .build();
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("authenticationTypeCode", getAuthenticationTypeCode())
+        .toString();
+  }
+}

polaris-core/src/test/java/org/apache/polaris/core/connection/ConnectionConfigInfoDpoTest.java

@@ -131,4 +131,42 @@ void testBearerAuthenticationParameters() throws JsonProcessingException {
         objectMapper.readValue(expectedApiModelJson, ConnectionConfigInfo.class),
         connectionConfigInfoApiModel);
   }
+
+  @Test
+  void testImplicitAuthenticationParameters() throws JsonProcessingException {
+    // Test deserialization and reserialization of the persistence JSON.
+    String json =
+        ""
+            + "{"
+            + "  \"connectionTypeCode\": 2,"
+            + "  \"uri\": \"file:///hadoop-catalog/warehouse\","
+            + "  \"warehouse\": \"hadoop-catalog\","
+            + "  \"authenticationParameters\": {"
+            + "    \"authenticationTypeCode\": 3"
+            + "  }"
+            + "}";
+    ConnectionConfigInfoDpo connectionConfigInfoDpo =
+        ConnectionConfigInfoDpo.deserialize(polarisDiagnostics, json);
+    Assertions.assertNotNull(connectionConfigInfoDpo);
+    JsonNode tree1 = objectMapper.readTree(json);
+    JsonNode tree2 = objectMapper.readTree(connectionConfigInfoDpo.serialize());
+    Assertions.assertEquals(tree1, tree2);
+
+    // Test conversion into API model JSON.
+    ConnectionConfigInfo connectionConfigInfoApiModel =
+        connectionConfigInfoDpo.asConnectionConfigInfoModel();
+    String expectedApiModelJson =
+        ""
+            + "{"
+            + "  \"connectionType\": \"HADOOP\","
+            + "  \"uri\": \"file:///hadoop-catalog/warehouse\","
+            + "  \"warehouse\": \"hadoop-catalog\","
+            + "  \"authenticationParameters\": {"
+            + "    \"authenticationType\": \"IMPLICIT\""
+            + "  }"
+            + "}";
+    Assertions.assertEquals(
+        objectMapper.readValue(expectedApiModelJson, ConnectionConfigInfo.class),
+        connectionConfigInfoApiModel);
+  }
 }

service/common/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java

@@ -19,6 +19,7 @@
 package org.apache.polaris.service.admin;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
 
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
@@ -29,6 +30,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -699,16 +701,10 @@ private Map<String, UserSecretReference> extractSecretReferences(
   /**
    * @see #extractSecretReferences
    */
-  private boolean requiresSecretReferenceExtraction(CreateCatalogRequest catalogRequest) {
-    Catalog catalog = catalogRequest.getCatalog();
-    if (catalog instanceof ExternalCatalog externalCatalog) {
-      if (externalCatalog.getConnectionConfigInfo() != null) {
-        // TODO: Make this more targeted once we have connection configs that don't involve
-        // processing of inline secrets.
-        return true;
-      }
-    }
-    return false;
+  private boolean requiresSecretReferenceExtraction(
+      @NotNull ConnectionConfigInfo connectionConfigInfo) {
+    return connectionConfigInfo.getAuthenticationParameters().getAuthenticationType()
+        != AuthenticationParameters.AuthenticationTypeEnum.IMPLICIT;
   }
 
   public PolarisEntity createCatalog(CreateCatalogRequest catalogRequest) {
@@ -733,24 +729,54 @@ public PolarisEntity createCatalog(CreateCatalogRequest catalogRequest) {
             .setProperties(reservedProperties.removeReservedProperties(entity.getPropertiesAsMap()))
             .build();
 
-    if (requiresSecretReferenceExtraction(catalogRequest)) {
-      LOGGER
-          .atDebug()
-          .addKeyValue("catalogName", entity.getName())
-          .log("Extracting secret references to create federated catalog");
-      FeatureConfiguration.enforceFeatureEnabledOrThrow(
-          callContext, FeatureConfiguration.ENABLE_CATALOG_FEDERATION);
-      // For fields that contain references to secrets, we'll separately process the secrets from
-      // the original request first, and then populate those fields with the extracted secret
-      // references as part of the construction of the internal persistence entity.
-      Map<String, UserSecretReference> processedSecretReferences =
-          extractSecretReferences(catalogRequest, entity);
-      entity =
-          new CatalogEntity.Builder(entity)
-              .setConnectionConfigInfoDpoWithSecrets(
-                  ((ExternalCatalog) catalogRequest.getCatalog()).getConnectionConfigInfo(),
-                  processedSecretReferences)
-              .build();
+    Catalog catalog = catalogRequest.getCatalog();
+    if (catalog instanceof ExternalCatalog externalCatalog) {
+      ConnectionConfigInfo connectionConfigInfo = externalCatalog.getConnectionConfigInfo();
+
+      if (connectionConfigInfo != null) {
+        LOGGER
+            .atDebug()
+            .addKeyValue("catalogName", entity.getName())
+            .log("Creating a federated catalog");
+        FeatureConfiguration.enforceFeatureEnabledOrThrow(
+            callContext, FeatureConfiguration.ENABLE_CATALOG_FEDERATION);
+        Map<String, UserSecretReference> processedSecretReferences = Map.of();
+        List<String> supportedAuthenticationTypes =
+            callContext
+                .getPolarisCallContext()
+                .getConfigurationStore()
+                .getConfiguration(
+                    callContext.getRealmContext(),
+                    FeatureConfiguration.SUPPORTED_EXTERNAL_CATALOG_AUTHENTICATION_TYPES)
+                .stream()
+                .map(s -> s.toUpperCase(Locale.ROOT))
+                .toList();
+        if (requiresSecretReferenceExtraction(connectionConfigInfo)) {
+          // For fields that contain references to secrets, we'll separately process the secrets
+          // from the original request first, and then populate those fields with the extracted
+          // secret references as part of the construction of the internal persistence entity.
+          checkState(
+              supportedAuthenticationTypes.contains(
+                  connectionConfigInfo
+                      .getAuthenticationParameters()
+                      .getAuthenticationType()
+                      .name()),
+              "Authentication type %s is not supported.",
+              connectionConfigInfo.getAuthenticationParameters().getAuthenticationType());
+          processedSecretReferences = extractSecretReferences(catalogRequest, entity);
+        } else {
+          // Support no-auth catalog federation only when the feature is enabled.
+          checkState(
+              supportedAuthenticationTypes.contains(
+                  AuthenticationParameters.AuthenticationTypeEnum.IMPLICIT.name()),
+              "Implicit authentication based catalog federation is not supported.");
+        }
+        entity =
+            new CatalogEntity.Builder(entity)
+                .setConnectionConfigInfoDpoWithSecrets(
+                    connectionConfigInfo, processedSecretReferences)
+                .build();
+      }
     }
 
     CreateCatalogResult catalogResult =

spec/polaris-management-service.yml

@@ -917,6 +917,7 @@ components:
             - OAUTH
             - BEARER
             - SIGV4
+            - IMPLICIT
           description: The type of authentication to use when connecting to the remote rest service
       required:
         - authenticationType
@@ -926,6 +927,7 @@ components:
           OAUTH: "#/components/schemas/OAuthClientCredentialsParameters"
           BEARER: "#/components/schemas/BearerAuthenticationParameters"
           SIGV4: "#/components/schemas/SigV4AuthenticationParameters"
+          IMPLICIT: "#/components/schemas/ImplicitAuthenticationParameters"
 
     OAuthClientCredentialsParameters:
       type: object
@@ -990,6 +992,13 @@ components:
         - roleArn
         - signingRegion
 
+    ImplicitAuthenticationParameters:
+      type: object
+      description: Polaris does not explicity accept any authentication parameters for the connection. Authentication 
+        parameters found in the environment and/or configuration files will be used for this connection.
+      allOf:
+        - $ref: '#/components/schemas/AuthenticationParameters'
+
     StorageConfigInfo:
       type: object
       description: A storage configuration used by catalogs