Initial MVP implementation of Catalog Federation to remote Iceberg REST Catalogs (#1305)

* Initial prototype of catalog federation just passing special properties into internal properties.

Make Resolver federation-aware to properly handle "best-effort" resolution of
passthrough facade entities.

Targets will automatically reflect the longest-path that we happen to have stored
locally and resolve grants against that path (including the degenerate case
where the longest-path is just the catalog itself).

This provides Catalog-level RBAC for passthrough federation.

Sketch out persistence-layer flow for how connection secrets might be pushed
down into a secrets-management layer.

* Defined internal representation classes for connection config

* Construct and initialize federated iceberg catalog based on connection config

* Apply the same spec renames to the internal ConnectionConfiguration representations.

* Manually pick @XJDKC fixes for integration tests and omittign secrets in response objects

* Fix internal connection structs with updated naming from spec PR

* Push CreateCatalogRequest down to PolarisAdminService::createCatalog just like UpdateCatalogRequest in updateCatalog.

This is needed if we're going to make PolarisAdminService handle secrets management without ever putting the secrets
into a CatalogEntity.

* Add new interface UserSecretsManager along with a default implementation

The default UnsafeInMemorySecretsManager just uses an inmemory ConcurrentHashMap
to store secrets, but structurally illustrates the full flow of intended
implementations.

For mutual protection against a compromise of a secret store or the core
persistence store, the default implementation demonstrates storing only
an encrypted secret in the secret store, and a one-time-pad key in the
returned referencePayload; other implementations using standard crypto
protocols may choose to instead only utilize the remote secret store as
the encryption keystore while storing the ciphertext in the referencePayload
(e.g. using a KMS engine with Vault vs using a KV engine).

Additionally, it demonstrates the use of an integrity check by storing a
basic hashCode in the referencePayload as well.

* Wire in UserSecretsManager to createCatalog and federated Iceberg API handlers

Update the internal DPOs corresponding to the various ConnectionConfigInfo API objects
to no longer contain any possible fields for inline secrets, instead holding the
JSON-serializable UserSecretReference corresponding to external/offloaded secrets.

CreateCatalog for federated catalogs containing secrets will now first extract
UserSecretReferences from the CreateCatalogRequest, and the CatalogEntity will
populate the DPOs corresponding to ConnectionConfigInfos in a secondary pass
by pulling out the relevant extracted UserSecretReferences.

For federated catalog requests, when reconstituting the actual sensitive
secret configs, the UserSecretsManager will be used to obtain the secrets
by using the stored UserSecretReferences.

Remove vestigial internal properties from earlier prototypes.

* Since we already use commons-codec DigestUtils.sha256Hex, use that for the hash in UnsafeInMemorySecretsManager
just for consistency and to illustrate a typical scenario using a cryptographic hash.

* Rename the persistence-objects corresponding to API model objects with a new naming
convention that just takes the API model object name and appends "Dpo" as a suffix;

* Use UserSecretsManagerFactory to Produce the UserSecretsManager (#1)

* Move PolarisAuthenticationParameters to a top-level property according to the latest spec

* Create a Factory for UserSecretsManager

* Fix a typo in UnsafeInMemorySecretsManagerFactory

* Gate all federation logic behind a new FeatureConfiguration - ENABLE_CATALOG_FEDERATION

* Also rename some variables and method names to be consistent with prior rename to ConnectionConfigInfoDpo

* Change ConnectionType and AuthenticationType to be stored as int codes in persistence objects.

Address PR feedback for various nits and javadoc comments.

* Add javadoc comment to IcebergCatalogPropertiesProvider

* Add some constraints on the expected format of the URN in UserSecretReference and placeholders
for next steps where we'd provide a ResolvingUserSecretsManager for example if the runtime ever
needs to delegate to two different implementations of UserSecretsManager for different entities.

Reduce the `forEntity` argument to just PolarisEntityCore to make it more clear that the
implementation is supposed to extract the necessary identifier info from forEntity for
backend cleanup and tracking purposes.

---------

Co-authored-by: Rulin Xing <rulin.xing+oss@snowflake.com>
Co-authored-by: Rulin Xing <xjdkcsq3@gmail.com>

Author: dennishuo

integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisApplicationIntegrationTest.java

@@ -378,7 +378,7 @@ public void testIcebergCreateTablesInExternalCatalog() throws IOException {
                       .withPartitionSpec(PartitionSpec.unpartitioned())
                       .create())
           .isInstanceOf(BadRequestException.class)
-          .hasMessage("Malformed request: Cannot create table on external catalogs.");
+          .hasMessage("Malformed request: Cannot create table on static-facade external catalogs.");
     }
   }
 
@@ -515,7 +515,7 @@ public void testIcebergUpdateTableInExternalCatalog() throws IOException {
                               10L))
                       .commit())
           .isInstanceOf(BadRequestException.class)
-          .hasMessage("Malformed request: Cannot update table on external catalogs.");
+          .hasMessage("Malformed request: Cannot update table on static-facade external catalogs.");
     }
   }
 

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

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.Optional;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.persistence.cache.EntityWeigher;
 
 /**
@@ -36,6 +37,22 @@ protected FeatureConfiguration(
     super(key, description, defaultValue, catalogConfig);
   }
 
+  /**
+   * Helper for the common scenario of gating a feature with a boolean FeatureConfiguration, where
+   * we want to throw an UnsupportedOperationException if it's not enabled.
+   */
+  public static void enforceFeatureEnabledOrThrow(
+      CallContext callContext, FeatureConfiguration<Boolean> featureConfig) {
+    boolean enabled =
+        callContext
+            .getPolarisCallContext()
+            .getConfigurationStore()
+            .getConfiguration(callContext.getPolarisCallContext(), featureConfig);
+    if (!enabled) {
+      throw new UnsupportedOperationException("Feature not enabled: " + featureConfig.key);
+    }
+  }
+
   public static final FeatureConfiguration<Boolean>
       ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING =
           PolarisConfiguration.<Boolean>builder()
@@ -201,4 +218,13 @@ protected FeatureConfiguration(
                   + " requires experimentation in the specific deployment environment")
           .defaultValue(100 * EntityWeigher.WEIGHT_PER_MB)
           .buildFeatureConfiguration();
+
+  public static final FeatureConfiguration<Boolean> ENABLE_CATALOG_FEDERATION =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ENABLE_CATALOG_FEDERATION")
+          .description(
+              "If true, allows creating and using ExternalCatalogs containing ConnectionConfigInfos"
+                  + " to perform federation to remote catalogs.")
+          .defaultValue(false)
+          .buildFeatureConfiguration();
 }

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

@@ -0,0 +1,88 @@
+/*
+ * 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.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+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.secrets.UserSecretReference;
+
+/**
+ * The internal persistence-object counterpart to AuthenticationParameters defined in the API model.
+ * Important: JsonSubTypes must be kept in sync with {@link AuthenticationType}.
+ */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "authenticationTypeCode", visible = true)
+@JsonSubTypes({
+  @JsonSubTypes.Type(value = OAuthClientCredentialsParametersDpo.class, name = "1"),
+  @JsonSubTypes.Type(value = BearerAuthenticationParametersDpo.class, name = "2"),
+})
+public abstract class AuthenticationParametersDpo implements IcebergCatalogPropertiesProvider {
+
+  public static final String INLINE_CLIENT_SECRET_REFERENCE_KEY = "inlineClientSecretReference";
+  public static final String INLINE_BEARER_TOKEN_REFERENCE_KEY = "inlineBearerTokenReference";
+
+  @JsonProperty(value = "authenticationTypeCode")
+  private final int authenticationTypeCode;
+
+  public AuthenticationParametersDpo(
+      @JsonProperty(value = "authenticationTypeCode", required = true) int authenticationTypeCode) {
+    this.authenticationTypeCode = authenticationTypeCode;
+  }
+
+  public int getAuthenticationTypeCode() {
+    return authenticationTypeCode;
+  }
+
+  public abstract AuthenticationParameters asAuthenticationParametersModel();
+
+  public static AuthenticationParametersDpo fromAuthenticationParametersModelWithSecrets(
+      AuthenticationParameters authenticationParameters,
+      Map<String, UserSecretReference> secretReferences) {
+    final AuthenticationParametersDpo config;
+    switch (authenticationParameters.getAuthenticationType()) {
+      case OAUTH:
+        OAuthClientCredentialsParameters oauthClientCredentialsModel =
+            (OAuthClientCredentialsParameters) authenticationParameters;
+        config =
+            new OAuthClientCredentialsParametersDpo(
+                AuthenticationType.OAUTH.getCode(),
+                oauthClientCredentialsModel.getTokenUri(),
+                oauthClientCredentialsModel.getClientId(),
+                secretReferences.get(INLINE_CLIENT_SECRET_REFERENCE_KEY),
+                oauthClientCredentialsModel.getScopes());
+        break;
+      case BEARER:
+        BearerAuthenticationParameters bearerAuthenticationParametersModel =
+            (BearerAuthenticationParameters) authenticationParameters;
+        config =
+            new BearerAuthenticationParametersDpo(
+                AuthenticationType.BEARER.getCode(),
+                secretReferences.get(INLINE_BEARER_TOKEN_REFERENCE_KEY));
+        break;
+      default:
+        throw new IllegalStateException(
+            "Unsupported authentication type: " + authenticationParameters.getAuthenticationType());
+    }
+    return config;
+  }
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+/**
+ * The internal persistence-object counterpart to AuthenticationParameters.AuthenticationTypeEnum
+ * defined in the API model. We define integer type codes in this enum for better compatibility
+ * within persisted data in case the names of enum types are ever changed in place.
+ *
+ * <p>Important: Codes must be kept in-sync with JsonSubTypes annotated within {@link
+ * AuthenticationParametersDpo}.
+ */
+public enum AuthenticationType {
+  OAUTH(1),
+  BEARER(2);
+
+  private final int code;
+
+  AuthenticationType(int code) {
+    this.code = code;
+  }
+
+  public int getCode() {
+    return this.code;
+  }
+}

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

@@ -0,0 +1,73 @@
+/*
+ * 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 jakarta.annotation.Nonnull;
+import java.util.Map;
+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.secrets.UserSecretReference;
+import org.apache.polaris.core.secrets.UserSecretsManager;
+
+/**
+ * The internal persistence-object counterpart to BearerAuthenticationParameters defined in the API
+ * model.
+ */
+public class BearerAuthenticationParametersDpo extends AuthenticationParametersDpo {
+
+  @JsonProperty(value = "bearerTokenReference")
+  private final UserSecretReference bearerTokenReference;
+
+  public BearerAuthenticationParametersDpo(
+      @JsonProperty(value = "authenticationTypeCode", required = true) int authenticationTypeCode,
+      @JsonProperty(value = "bearerTokenReference", required = true) @Nonnull
+          UserSecretReference bearerTokenReference) {
+    super(authenticationTypeCode);
+    this.bearerTokenReference = bearerTokenReference;
+  }
+
+  public @Nonnull UserSecretReference getBearerTokenReference() {
+    return bearerTokenReference;
+  }
+
+  @Override
+  public @Nonnull Map<String, String> asIcebergCatalogProperties(
+      UserSecretsManager secretsManager) {
+    String bearerToken = secretsManager.readSecret(getBearerTokenReference());
+    return Map.of(OAuth2Properties.TOKEN, bearerToken);
+  }
+
+  @Override
+  public AuthenticationParameters asAuthenticationParametersModel() {
+    return BearerAuthenticationParameters.builder()
+        .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.BEARER)
+        .build();
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("authenticationTypeCode", getAuthenticationTypeCode())
+        .add("bearerTokenReference", getBearerTokenReference())
+        .toString();
+  }
+}