[Cherry-pick to branch-1.3] [#14069] fix(core): Reject token credential providers on catalogs without path-based credentials (#14080)

catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogCredential.java

@@ -18,6 +18,7 @@
  */
 package org.apache.gravitino.catalog.fileset;
 
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import java.time.Instant;
 import java.util.Map;
@@ -74,4 +75,14 @@ void testLocationPrefixNotUserConfigurable() {
             .propertiesWithCredentialProviders()
             .get(CredentialConstants.S3_CREDENTIAL_LIST_LOCATION_PREFIX));
   }
+
+  @Test
+  void testPathBasedCredentialProvidersAllowed() {
+    FilesetCatalogImpl catalog = newCatalog(Maps.newHashMap());
+    Assertions.assertTrue(catalog.supportsPathBasedCredentials());
+    Assertions.assertDoesNotThrow(
+        () ->
+            catalog.validateCredentialProviders(
+                ImmutableMap.of(CredentialConstants.CREDENTIAL_PROVIDERS, "s3-token,gcs-token")));
+  }
 }

catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalog.java

@@ -54,6 +54,11 @@ public String shortName() {
     return "glue";
   }
 
+  @Override
+  public boolean supportsPathBasedCredentials() {
+    return false;
+  }
+
   /**
    * Creates a new instance of {@link GlueCatalogOperations} with the provided configuration.
    *

catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalog.java

@@ -51,6 +51,11 @@ public String shortName() {
     return "hive";
   }
 
+  @Override
+  public boolean supportsPathBasedCredentials() {
+    return false;
+  }
+
   /**
    * Creates a new instance of {@link HiveCatalogOperations} with the provided configuration.
    *

catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestHiveCatalog.java

@@ -38,6 +38,7 @@
 import org.apache.gravitino.connector.PropertiesMetadata;
 import org.apache.gravitino.connector.capability.Capability;
 import org.apache.gravitino.connector.capability.CapabilityResult;
+import org.apache.gravitino.credential.CredentialConstants;
 import org.apache.gravitino.hive.client.HiveClientClassLoader.HiveVersion;
 import org.apache.gravitino.hive.hms.MiniHiveMetastoreService;
 import org.apache.gravitino.meta.AuditInfo;
@@ -228,4 +229,28 @@ void testCatalogProperty() {
                     "Properties or property prefixes are required and must be set: [%s]",
                     METASTORE_URIS)));
   }
+
+  @Test
+  void testPathBasedCredentialProvidersRejected() {
+    CatalogEntity entity =
+        CatalogEntity.builder()
+            .withId(1L)
+            .withName("catalog")
+            .withNamespace(Namespace.of("metalake"))
+            .withType(HiveCatalog.Type.RELATIONAL)
+            .withProvider("hive")
+            .withAuditInfo(
+                AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build())
+            .build();
+    HiveCatalog catalog = new HiveCatalog().withCatalogEntity(entity);
+
+    Assertions.assertFalse(catalog.supportsPathBasedCredentials());
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                catalog.validateCredentialProviders(
+                    ImmutableMap.of(CredentialConstants.CREDENTIAL_PROVIDERS, "s3-token")));
+    Assertions.assertTrue(exception.getMessage().contains("s3-token"));
+  }
 }

catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalog.java

@@ -81,6 +81,11 @@ public Capability newCapability() {
     return new JdbcCatalogCapability();
   }
 
+  @Override
+  public boolean supportsPathBasedCredentials() {
+    return false;
+  }
+
   /**
    * Returns the effective {@link JdbcConfig#VERIFY_ON_CREATE verify-on-create} setting for this
    * catalog. A relational JDBC catalog verifies its backend at create time by default; an

catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalog.java

@@ -42,6 +42,11 @@ public String shortName() {
     return "kafka";
   }
 
+  @Override
+  public boolean supportsPathBasedCredentials() {
+    return false;
+  }
+
   @Override
   protected CatalogOperations newOps(Map<String, String> config) {
     KafkaCatalogOperations ops = new KafkaCatalogOperations();

catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/generic/GenericCatalog.java

@@ -46,6 +46,11 @@ public String shortName() {
     return "lakehouse-generic";
   }
 
+  @Override
+  public boolean supportsPathBasedCredentials() {
+    return false;
+  }
+
   /**
    * Creates a new instance of {@link GenericCatalogOperations} with the provided configuration.
    *

catalogs/catalog-lakehouse-hudi/src/main/java/org/apache/gravitino/catalog/lakehouse/hudi/HudiCatalog.java

@@ -43,6 +43,11 @@ public String shortName() {
     return "lakehouse-hudi";
   }
 
+  @Override
+  public boolean supportsPathBasedCredentials() {
+    return false;
+  }
+
   /**
    * Creates a new instance of {@link HudiCatalogOperations} with the provided configuration.
    *

catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalog.java

@@ -22,6 +22,7 @@
 import static org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalog.SCHEMA_PROPERTIES_META;
 import static org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalog.TABLE_PROPERTIES_META;
 
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import java.time.Instant;
 import java.util.Map;
@@ -490,4 +491,14 @@ void testJdbcBackendWithAzureCredentialProviders() {
     Assertions.assertTrue(
         credentialProviders.contains(AzureAccountKeyCredential.AZURE_ACCOUNT_KEY_CREDENTIAL_TYPE));
   }
+
+  @Test
+  void testPathBasedCredentialProvidersAllowed() {
+    IcebergCatalog catalog = new IcebergCatalog();
+    Assertions.assertTrue(catalog.supportsPathBasedCredentials());
+    Assertions.assertDoesNotThrow(
+        () ->
+            catalog.validateCredentialProviders(
+                ImmutableMap.of(CredentialConstants.CREDENTIAL_PROVIDERS, "s3-token,gcs-token")));
+  }
 }

catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalog.java

@@ -51,6 +51,11 @@ public String shortName() {
     return "lakehouse-paimon";
   }
 
+  @Override
+  public boolean supportsPathBasedCredentials() {
+    return false;
+  }
+
   /**
    * Creates a new instance of {@link PaimonCatalogOperations} with the provided configuration.
    *

catalogs/catalog-model/src/main/java/org/apache/gravitino/catalog/model/ModelCatalogImpl.java

@@ -45,6 +45,11 @@ public String shortName() {
     return "model";
   }
 
+  @Override
+  public boolean supportsPathBasedCredentials() {
+    return false;
+  }
+
   @Override
   protected CatalogOperations newOps(Map<String, String> config) {
     EntityStore store = GravitinoEnv.getInstance().entityStore();

core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java

@@ -1023,6 +1023,7 @@ public void testConnection(NameIdentifier ident, CatalogChange... changes) {
                       metadata.catalogPropertiesMetadata(),
                       alterProperty.getLeft(),
                       alterProperty.getRight());
+                  storedCatalog.validateCredentialProviders(alterProperty.getLeft());
                   return null;
                 });
 
@@ -1182,6 +1183,7 @@ public Catalog alterCatalog(NameIdentifier ident, CatalogChange... changes)
                             f.catalogPropertiesMetadata(),
                             alterProperty.getLeft(),
                             alterProperty.getRight());
+                        catalog.validateCredentialProviders(alterProperty.getLeft());
                         return null;
                       });
             } catch (IllegalArgumentException e1) {
@@ -1705,6 +1707,19 @@ CatalogWrapper createCatalogWrapper(
     classLoader.withClassLoader(
         cl -> {
           validatePropertyForCreate(catalog.catalogPropertiesMetadata(), propsToValidate);
+          catalog.validateCredentialProviders(propsToValidate);
+          if (propsToValidate == null) {
+            // Loading stays lenient for catalogs persisted with unsupported providers, so only
+            // warn about them.
+            try {
+              catalog.validateCredentialProviders(conf);
+            } catch (IllegalArgumentException e) {
+              LOG.warn(
+                  "Catalog {} has credential providers that cannot vend credentials: {}",
+                  entity.nameIdentifier(),
+                  e.getMessage());
+            }
+          }
 
           // Call wrapper.catalog.properties() to make BaseCatalog#properties in IsolatedClassLoader
           // not null. Why do we do this? Because wrapper.catalog.properties() needs to be called in

core/src/main/java/org/apache/gravitino/connector/BaseCatalog.java

@@ -26,8 +26,10 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.gravitino.Audit;
@@ -40,9 +42,11 @@
 import org.apache.gravitino.connector.authorization.AuthorizationPlugin;
 import org.apache.gravitino.connector.authorization.BaseAuthorization;
 import org.apache.gravitino.connector.capability.Capability;
+import org.apache.gravitino.connector.credential.SupportsPathBasedCredentials;
 import org.apache.gravitino.credential.AzureAccountKeyCredential;
 import org.apache.gravitino.credential.CatalogCredentialManager;
 import org.apache.gravitino.credential.CredentialConstants;
+import org.apache.gravitino.credential.CredentialUtils;
 import org.apache.gravitino.credential.GCSTokenCredential;
 import org.apache.gravitino.credential.OSSSecretKeyCredential;
 import org.apache.gravitino.credential.S3SecretKeyCredential;
@@ -237,6 +241,50 @@ public void validateOnCreate() {
     ops();
   }
 
+  /**
+   * Whether this catalog provides path-based credential contexts, which token credential providers
+   * such as {@code s3-token} require to vend credentials.
+   *
+   * <p>A catalog that cannot provide path-based contexts, neither through operations implementing
+   * {@link SupportsPathBasedCredentials} nor through another server such as the Iceberg REST
+   * server, should override this method to return {@code false}, so that creating or altering it
+   * with a token provider fails.
+   *
+   * @return {@code true} (default) if path-based credentials are supported; {@code false} otherwise
+   */
+  public boolean supportsPathBasedCredentials() {
+    return true;
+  }
+
+  /**
+   * Validates that the credential providers set in {@code properties} can vend credentials for this
+   * catalog. Token providers listed in {@link CredentialUtils#PATH_BASED_CREDENTIAL_TYPES} are
+   * rejected when {@link #supportsPathBasedCredentials()} returns {@code false}.
+   *
+   * @param properties the catalog properties to validate, may be null
+   * @throws IllegalArgumentException if a credential provider cannot be vended by this catalog
+   */
+  public void validateCredentialProviders(@Nullable Map<String, String> properties) {
+    if (properties == null || supportsPathBasedCredentials()) {
+      return;
+    }
+    List<String> unsupportedProviders =
+        CredentialUtils.getCredentialProvidersByOrder(() -> properties).stream()
+            // Provider lookup ignores case, so match the type names case-insensitively too.
+            .filter(
+                p ->
+                    CredentialUtils.PATH_BASED_CREDENTIAL_TYPES.contains(
+                        p.toLowerCase(Locale.ROOT)))
+            .sorted()
+            .collect(Collectors.toList());
+    Preconditions.checkArgument(
+        unsupportedProviders.isEmpty(),
+        "Credential providers %s are not supported by catalog provider %s, it cannot vend"
+            + " path-based credentials",
+        unsupportedProviders,
+        provider());
+  }
+
   public void checkMetalakeAndCatalogInUse() {
     Map<String, String> catalogProperties = entity.getProperties();
     if (catalogProperties == null) {
@@ -587,7 +635,8 @@ protected void addStorageCredentialProviders(
     }
 
     String gcsServiceAccountFile = properties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE);
-    if (StringUtils.isNotBlank(gcsServiceAccountFile)) {
+    // gcs-token only vends credentials for path-based contexts.
+    if (StringUtils.isNotBlank(gcsServiceAccountFile) && supportsPathBasedCredentials()) {
       credentialProviders.add(GCSTokenCredential.GCS_TOKEN_CREDENTIAL_TYPE);
     }
   }

core/src/main/java/org/apache/gravitino/credential/CredentialUtils.java

@@ -19,6 +19,7 @@
 
 package org.apache.gravitino.credential;
 
+import com.google.common.collect.ImmutableSet;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -29,6 +30,18 @@
 
 public class CredentialUtils {
 
+  /**
+   * Credential types that only vend credentials for a path-based credential context. They return
+   * nothing for a catalog-level context, so they only work on catalogs that provide path contexts.
+   */
+  public static final Set<String> PATH_BASED_CREDENTIAL_TYPES =
+      ImmutableSet.of(
+          S3TokenCredential.S3_TOKEN_CREDENTIAL_TYPE,
+          AwsIrsaCredential.AWS_IRSA_CREDENTIAL_TYPE,
+          OSSTokenCredential.OSS_TOKEN_CREDENTIAL_TYPE,
+          GCSTokenCredential.GCS_TOKEN_CREDENTIAL_TYPE,
+          ADLSTokenCredential.ADLS_TOKEN_CREDENTIAL_TYPE);
+
   public static Map<String, CredentialProvider> loadCredentialProviders(
       Map<String, String> catalogProperties) {
     CredentialConfig credentialConfig = new CredentialConfig(catalogProperties);

core/src/test/java/org/apache/gravitino/TestCatalog.java

@@ -55,6 +55,11 @@ public String shortName() {
     return "test";
   }
 
+  @Override
+  public boolean supportsPathBasedCredentials() {
+    return false;
+  }
+
   @Override
   protected CatalogOperations newOps(Map<String, String> config) {
     return new TestCatalogOperations(config);