Improve Access Delegation Mode Selection Algorithm

runtime/service/src/main/java/org/apache/polaris/service/catalog/AccessDelegationModeResolver.java

@@ -0,0 +1,68 @@
+/*
+ * 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.service.catalog;
+
+import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
+import java.util.EnumSet;
+import org.apache.polaris.core.entity.CatalogEntity;
+import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+
+/**
+ * Resolves the optimal {@link AccessDelegationMode} to use based on the client's requested modes
+ * and the catalog's capabilities.
+ *
+ * <p>The selection algorithm is:
+ *
+ * <ol>
+ *   <li>If no delegation mode is requested, use {@link AccessDelegationMode#UNKNOWN}
+ *   <li>If exactly one delegation mode is requested, use that mode (if supported)
+ *   <li>If the requested modes include both {@link AccessDelegationMode#VENDED_CREDENTIALS} and
+ *       {@link AccessDelegationMode#REMOTE_SIGNING}:
+ *       <ol>
+ *         <li>Check if STS is available for the catalog's storage configuration
+ *         <li>If STS is available and credential subscoping is not skipped, use {@link
+ *             AccessDelegationMode#VENDED_CREDENTIALS}
+ *         <li>Otherwise, use {@link AccessDelegationMode#REMOTE_SIGNING}
+ *       </ol>
+ * </ol>
+ *
+ * <p>This resolver improves upon the simple mode selection by considering:
+ *
+ * <ul>
+ *   <li>STS availability from the catalog's {@link AwsStorageConfigurationInfo}
+ *   <li>Feature configuration settings for credential subscoping
+ * </ul>
+ */
+public interface AccessDelegationModeResolver {
+
+  /**
+   * Resolves the optimal access delegation mode based on the requested modes and catalog
+   * capabilities.
+   *
+   * @param requestedModes The set of delegation modes requested by the client
+   * @param catalogEntity The catalog entity, used to determine storage configuration and
+   *     capabilities
+   * @return The resolved access delegation mode
+   */
+  @Nonnull
+  AccessDelegationMode resolve(
+      @Nonnull EnumSet<AccessDelegationMode> requestedModes,
+      @Nullable CatalogEntity catalogEntity);
+}

runtime/service/src/main/java/org/apache/polaris/service/catalog/DefaultAccessDelegationModeResolver.java

@@ -0,0 +1,220 @@
+/*
+ * 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.service.catalog;
+
+import static org.apache.polaris.service.catalog.AccessDelegationMode.REMOTE_SIGNING;
+import static org.apache.polaris.service.catalog.AccessDelegationMode.UNKNOWN;
+import static org.apache.polaris.service.catalog.AccessDelegationMode.VENDED_CREDENTIALS;
+
+import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
+import jakarta.enterprise.context.RequestScoped;
+import jakarta.inject.Inject;
+import java.util.EnumSet;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.config.PolarisConfigurationStore;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.entity.CatalogEntity;
+import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
+import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Default implementation of {@link AccessDelegationModeResolver} that resolves the optimal access
+ * delegation mode based on catalog capabilities and configuration.
+ *
+ * <p>The resolution logic:
+ *
+ * <ol>
+ *   <li>If no delegation mode is requested, returns {@link AccessDelegationMode#UNKNOWN}
+ *   <li>If exactly one delegation mode is requested (excluding UNKNOWN), returns that mode
+ *   <li>If both {@link AccessDelegationMode#VENDED_CREDENTIALS} and {@link
+ *       AccessDelegationMode#REMOTE_SIGNING} are requested:
+ *       <ul>
+ *         <li>If STS is unavailable for the catalog's AWS storage, returns {@link
+ *             AccessDelegationMode#REMOTE_SIGNING}
+ *         <li>If credential subscoping is skipped, returns {@link
+ *             AccessDelegationMode#REMOTE_SIGNING}
+ *         <li>Otherwise, returns {@link AccessDelegationMode#VENDED_CREDENTIALS}
+ *       </ul>
+ * </ol>
+ */
+@RequestScoped
+public class DefaultAccessDelegationModeResolver implements AccessDelegationModeResolver {
+
+  private static final Logger LOGGER =
+      LoggerFactory.getLogger(DefaultAccessDelegationModeResolver.class);
+
+  private final PolarisConfigurationStore configurationStore;
+  private final RealmContext realmContext;
+
+  @Inject
+  public DefaultAccessDelegationModeResolver(
+      PolarisConfigurationStore configurationStore, RealmContext realmContext) {
+    this.configurationStore = configurationStore;
+    this.realmContext = realmContext;
+  }
+
+  @Override
+  @Nonnull
+  public AccessDelegationMode resolve(
+      @Nonnull EnumSet<AccessDelegationMode> requestedModes,
+      @Nullable CatalogEntity catalogEntity) {
+
+    // Case 1: No delegation mode requested
+    if (requestedModes.isEmpty()) {
+      LOGGER.debug("No delegation mode requested, returning UNKNOWN");
+      return UNKNOWN;
+    }
+
+    // Filter out UNKNOWN mode from consideration for selection logic
+    EnumSet<AccessDelegationMode> effectiveModes = EnumSet.copyOf(requestedModes);
+    effectiveModes.remove(UNKNOWN);
+
+    if (effectiveModes.isEmpty()) {
+      LOGGER.debug("Only UNKNOWN mode requested, returning UNKNOWN");
+      return UNKNOWN;
+    }
+
+    // Case 2: Exactly one delegation mode requested
+    if (effectiveModes.size() == 1) {
+      AccessDelegationMode mode = effectiveModes.iterator().next();
+      LOGGER.debug("Single delegation mode requested: {}", mode);
+      return mode;
+    }
+
+    // Case 3: Both VENDED_CREDENTIALS and REMOTE_SIGNING requested
+    if (effectiveModes.contains(VENDED_CREDENTIALS) && effectiveModes.contains(REMOTE_SIGNING)) {
+      return resolveVendedCredentialsVsRemoteSigning(catalogEntity);
+    }
+
+    // Case 4: Unknown combination - default to VENDED_CREDENTIALS for backward compatibility
+    LOGGER.warn(
+        "Unknown access delegation mode combination: {}, defaulting to VENDED_CREDENTIALS",
+        requestedModes);
+    return VENDED_CREDENTIALS;
+  }
+
+  /**
+   * Resolves between VENDED_CREDENTIALS and REMOTE_SIGNING based on catalog capabilities.
+   *
+   * <p>The logic prefers VENDED_CREDENTIALS when:
+   *
+   * <ul>
+   *   <li>STS is available for the catalog's storage (for AWS)
+   *   <li>Credential subscoping is not skipped
+   * </ul>
+   *
+   * <p>Otherwise, REMOTE_SIGNING is preferred as it doesn't require STS.
+   */
+  private AccessDelegationMode resolveVendedCredentialsVsRemoteSigning(
+      @Nullable CatalogEntity catalogEntity) {
+
+    // If no catalog entity available, default to VENDED_CREDENTIALS for backward compatibility
+    if (catalogEntity == null) {
+      LOGGER.debug(
+          "No catalog entity available for mode resolution, defaulting to VENDED_CREDENTIALS");
+      return VENDED_CREDENTIALS;
+    }
+
+    // Check if credential subscoping is authorized for this catalog.
+    // For internal catalogs, credential subscoping is always authorized.
+    // For external/federated catalogs, check if ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING is enabled.
+    boolean credentialSubscopingAuthorized =
+        !catalogEntity.isExternal()
+            || configurationStore.getConfiguration(
+                realmContext,
+                catalogEntity,
+                FeatureConfiguration.ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING);
+
+    if (!credentialSubscopingAuthorized) {
+      LOGGER.debug(
+          "Credential subscoping is not authorized for external catalog {}, selecting REMOTE_SIGNING",
+          catalogEntity.getName());
+      return REMOTE_SIGNING;
+    }
+
+    // Check if credential subscoping is skipped - if so, VENDED_CREDENTIALS won't work properly
+    // Note: This config is realm-level only (not overridable at catalog level) and has a default value
+    boolean skipCredentialSubscoping =
+        configurationStore.getConfiguration(
+            realmContext, FeatureConfiguration.SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION);
+
+    if (skipCredentialSubscoping) {
+      LOGGER.debug(
+          "Credential subscoping is skipped for this realm, selecting REMOTE_SIGNING");
+      return REMOTE_SIGNING;
+    }
+
+    // Check credential vending availability from storage configuration
+    boolean credentialVendingAvailable = isCredentialVendingAvailable(catalogEntity);
+
+    if (credentialVendingAvailable) {
+      LOGGER.debug(
+          "Credential vending is available for catalog {}, selecting VENDED_CREDENTIALS",
+          catalogEntity.getName());
+      return VENDED_CREDENTIALS;
+    } else {
+      LOGGER.debug(
+          "Credential vending is not available for catalog {}, selecting REMOTE_SIGNING",
+          catalogEntity.getName());
+      return REMOTE_SIGNING;
+    }
+  }
+
+  /**
+   * Determines if STS is available for the catalog's storage configuration.
+   *
+   * <p>For AWS storage, this checks the {@link AwsStorageConfigurationInfo#getStsUnavailable()}
+   * flag. For other storage types, STS is considered available by default (since they may have
+   * their own credential vending mechanisms).
+   *
+   * @param catalogEntity The catalog entity to check
+   * @return true if STS is available or the storage type doesn't require STS
+   */
+  private boolean isCredentialVendingAvailable(@Nonnull CatalogEntity catalogEntity) {
+    PolarisStorageConfigurationInfo storageConfig = catalogEntity.getStorageConfigurationInfo();
+
+    if (storageConfig == null) {
+      LOGGER.debug(
+          "No storage configuration found for catalog {}, assuming STS is available",
+          catalogEntity.getName());
+      return true;
+    }
+
+    if (storageConfig instanceof AwsStorageConfigurationInfo awsConfig) {
+      // STS is available unless explicitly marked as unavailable
+      boolean stsUnavailable = Boolean.TRUE.equals(awsConfig.getStsUnavailable());
+      LOGGER.debug(
+          "AWS storage configuration for catalog {}: stsUnavailable={}",
+          catalogEntity.getName(),
+          stsUnavailable);
+      return !stsUnavailable;
+    }
+
+    // For non-AWS storage types, assume STS-like functionality is available
+    // (e.g., Azure uses different mechanisms, GCP uses service accounts)
+    LOGGER.debug(
+        "Non-AWS storage type {} for catalog {}, assuming credential vending is available",
+        storageConfig.getStorageType(),
+        catalogEntity.getName());
+    return true;
+  }
+}

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogAdapter.java

@@ -18,6 +18,7 @@
  */
 package org.apache.polaris.service.catalog.iceberg;
 
+import static org.apache.polaris.service.catalog.AccessDelegationMode.REMOTE_SIGNING;
 import static org.apache.polaris.service.catalog.AccessDelegationMode.VENDED_CREDENTIALS;
 import static org.apache.polaris.service.catalog.validation.IcebergPropertiesValidation.validateIcebergProperties;
 
@@ -243,13 +244,16 @@ public Response updateProperties(
   }
 
   private EnumSet<AccessDelegationMode> parseAccessDelegationModes(String accessDelegationMode) {
-    EnumSet<AccessDelegationMode> delegationModes =
+    // Parse the modes from the header - validation will happen after mode resolution
+    // in IcebergCatalogHandler.resolveAccessDelegationModes()
+    EnumSet<AccessDelegationMode> modes =
         AccessDelegationMode.fromProtocolValuesList(accessDelegationMode);
+    // TODO remove when remote signing is implemented
     Preconditions.checkArgument(
-        delegationModes.isEmpty() || delegationModes.contains(VENDED_CREDENTIALS),
+        !modes.contains(REMOTE_SIGNING),
         "Unsupported access delegation mode: %s",
-        accessDelegationMode);
-    return delegationModes;
+        REMOTE_SIGNING);
+    return modes;
   }
 
   @Override