[#11263] feat(authz): Extend credential vending to Hive/Iceberg/Glue/JDBC catalog types

api/src/main/java/org/apache/gravitino/credential/JdbcCredential.java

@@ -107,8 +107,7 @@ public String toString() {
 
   private void validate(String jdbcUser, String jdbcPassword, long expireTimeInMs) {
     Preconditions.checkArgument(StringUtils.isNotBlank(jdbcUser), "JDBC user should not be empty");
-    Preconditions.checkArgument(
-        StringUtils.isNotBlank(jdbcPassword), "JDBC password should not be empty");
+    Preconditions.checkArgument(jdbcPassword != null, "JDBC password should not be null");
     // JDBC credentials are static (no server-issued expiry). expireTimeInMs must always be 0.
     Preconditions.checkArgument(
         expireTimeInMs == 0, "The expire time of JdbcCredential should be 0");

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

@@ -18,11 +18,14 @@
  */
 package org.apache.gravitino.catalog.glue;
 
+import java.util.List;
 import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.gravitino.connector.BaseCatalog;
 import org.apache.gravitino.connector.CatalogOperations;
 import org.apache.gravitino.connector.PropertiesMetadata;
 import org.apache.gravitino.connector.capability.Capability;
+import org.apache.gravitino.storage.S3Properties;
 
 /**
  * Implementation of an AWS Glue Data Catalog connector in Apache Gravitino.
@@ -81,4 +84,16 @@ public PropertiesMetadata schemaPropertiesMetadata() throws UnsupportedOperation
   public PropertiesMetadata tablePropertiesMetadata() throws UnsupportedOperationException {
     return TABLE_PROPERTIES_METADATA;
   }
+
+  @Override
+  protected void addCatalogSpecificCredentialProviders(
+      Map<String, String> properties, List<String> credentialProviders) {
+    String accessKeyId = properties.get(GlueConstants.AWS_ACCESS_KEY_ID);
+    String secretAccessKey = properties.get(GlueConstants.AWS_SECRET_ACCESS_KEY);
+    if (StringUtils.isNotBlank(accessKeyId) && StringUtils.isNotBlank(secretAccessKey)) {
+      properties.putIfAbsent(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID, accessKeyId);
+      properties.putIfAbsent(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY, secretAccessKey);
+    }
+    addStorageCredentialProviders(properties, credentialProviders);
+  }
 }

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

@@ -74,7 +74,7 @@ public class GlueCatalogPropertiesMetadata extends BaseCatalogPropertiesMetadata
                       + " When omitted the default credential chain is used.",
                   false /* immutable */,
                   null /* defaultValue */,
-                  false /* hidden */))
+                  true /* hidden */))
           .put(
               AWS_GLUE_ENDPOINT,
               stringOptionalPropertyEntry(

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

@@ -23,6 +23,10 @@
 
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
+import org.apache.gravitino.cloud.storage.AzurePropertiesMetadata;
+import org.apache.gravitino.cloud.storage.GCSPropertiesMetadata;
+import org.apache.gravitino.cloud.storage.OSSPropertiesMetadata;
+import org.apache.gravitino.cloud.storage.S3PropertiesMetadata;
 import org.apache.gravitino.connector.BaseCatalogPropertiesMetadata;
 import org.apache.gravitino.connector.PropertyEntry;
 import org.apache.gravitino.hive.ClientPropertiesMetadata;
@@ -123,6 +127,10 @@ public class HiveCatalogPropertiesMetadata extends BaseCatalogPropertiesMetadata
                   DEFAULT_LIST_ALL_TABLES,
                   false /* hidden */,
                   false /* reserved */))
+          .putAll(S3PropertiesMetadata.PROPERTY_ENTRIES)
+          .putAll(OSSPropertiesMetadata.PROPERTY_ENTRIES)
+          .putAll(AzurePropertiesMetadata.PROPERTY_ENTRIES)
+          .putAll(GCSPropertiesMetadata.PROPERTY_ENTRIES)
           .putAll(CLIENT_PROPERTIES_METADATA.propertyEntries())
           .build();
 

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

@@ -88,7 +88,7 @@ void testPropertyMeta() {
     Map<String, PropertyEntry<?>> propertyEntryMap =
         HIVE_PROPERTIES_METADATA.catalogPropertiesMetadata().propertyEntries();
 
-    Assertions.assertEquals(18, propertyEntryMap.size());
+    Assertions.assertEquals(25, propertyEntryMap.size());
     Assertions.assertTrue(propertyEntryMap.containsKey(METASTORE_URIS));
     Assertions.assertTrue(propertyEntryMap.containsKey(Catalog.PROPERTY_PACKAGE));
     Assertions.assertTrue(propertyEntryMap.containsKey(BaseCatalog.CATALOG_OPERATION_IMPL));

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

@@ -18,14 +18,10 @@
  */
 package org.apache.gravitino.catalog.jdbc;
 
-import com.google.common.collect.Maps;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.gravitino.Config;
-import org.apache.gravitino.Configs;
-import org.apache.gravitino.GravitinoEnv;
-import org.apache.gravitino.annotation.Evolving;
 import org.apache.gravitino.catalog.jdbc.config.JdbcConfig;
 import org.apache.gravitino.catalog.jdbc.converter.JdbcColumnDefaultValueConverter;
 import org.apache.gravitino.catalog.jdbc.converter.JdbcExceptionConverter;
@@ -37,7 +33,6 @@
 import org.apache.gravitino.connector.PropertiesMetadata;
 import org.apache.gravitino.connector.PropertyEntry;
 import org.apache.gravitino.connector.capability.Capability;
-import org.apache.gravitino.credential.CredentialConstants;
 import org.apache.gravitino.credential.JdbcCredential;
 
 /** Implementation of an Jdbc catalog in Gravitino. */
@@ -124,61 +119,12 @@ public PropertiesMetadata tablePropertiesMetadata() throws UnsupportedOperationE
   }
 
   @Override
-  @Evolving
-  public Map<String, String> propertiesWithCredentialProviders() {
-    // Use raw entity properties so that hidden credentials (jdbc-user/jdbc-password) are visible
-    // to the credential manager even after they are marked hidden in the properties metadata.
-    Map<String, String> properties = Maps.newHashMap(entity().getProperties());
-    return applyDefaultCredentialProviders(properties);
-  }
-
-  /**
-   * Returns catalog properties, optionally re-adding hidden JDBC credentials for backward
-   * compatibility with connectors that do not support credential vending. The backfill behavior is
-   * controlled by the server-level config {@code
-   * gravitino.catalog.credential.backfillToProperties}; it is disabled by default and should only
-   * be enabled during rolling upgrades.
-   *
-   * @return the catalog properties map, with credentials backfilled if the server config is set
-   */
-  @Override
-  public Map<String, String> properties() {
-    Map<String, String> props = super.properties();
-    if (!shouldBackfillCredential()) {
-      return props;
-    }
-    // Backfill hidden credentials for backward compatibility with connectors that do not support
-    // credential vending.
-    Map<String, String> rawProps = entity().getProperties();
-    Map<String, String> result = Maps.newHashMap(props);
-    String user = rawProps.get(JdbcConfig.USERNAME.getKey());
-    String password = rawProps.get(JdbcConfig.PASSWORD.getKey());
-    if (user != null) {
-      result.put(JdbcConfig.USERNAME.getKey(), user);
-    }
-    if (password != null) {
-      result.put(JdbcConfig.PASSWORD.getKey(), password);
-    }
-    return result;
-  }
-
-  private boolean shouldBackfillCredential() {
-    Config serverConfig = GravitinoEnv.getInstance().config();
-    return serverConfig != null
-        && serverConfig.get(Configs.CATALOG_CREDENTIAL_BACKFILL_TO_PROPERTIES);
-  }
-
-  private Map<String, String> applyDefaultCredentialProviders(Map<String, String> properties) {
-    if (StringUtils.isNotBlank(properties.get(CredentialConstants.CREDENTIAL_PROVIDERS))) {
-      return properties;
-    }
-
+  protected void addCatalogSpecificCredentialProviders(
+      Map<String, String> properties, List<String> credentialProviders) {
     String jdbcUser = properties.get(JdbcConfig.USERNAME.getKey());
     String jdbcPassword = properties.get(JdbcConfig.PASSWORD.getKey());
-    if (StringUtils.isNotBlank(jdbcUser) && StringUtils.isNotBlank(jdbcPassword)) {
-      properties.put(CredentialConstants.CREDENTIAL_PROVIDERS, JdbcCredential.JDBC_CREDENTIAL_TYPE);
+    if (StringUtils.isNotBlank(jdbcUser) && jdbcPassword != null) {
+      credentialProviders.add(JdbcCredential.JDBC_CREDENTIAL_TYPE);
     }
-
-    return properties;
   }
 }

catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogCredential.java

@@ -109,6 +109,38 @@ void testJdbcCatalogDefaultCredentialProviders() {
     Assertions.assertEquals(JdbcCredential.JDBC_CREDENTIAL_TYPE, credentialProviders);
   }
 
+  @Test
+  void testJdbcCatalogDefaultCredentialProvidersWithEmptyPassword() {
+    AuditInfo auditInfo =
+        AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+
+    // Empty-string password (e.g. StarRocks default) should still register the credential provider
+    Map<String, String> jdbcProps = Maps.newHashMap();
+    jdbcProps.put(JdbcConfig.JDBC_URL.getKey(), "jdbc:mysql://localhost:9030/");
+    jdbcProps.put(JdbcConfig.JDBC_DRIVER.getKey(), "com.mysql.cj.jdbc.Driver");
+    jdbcProps.put(JdbcConfig.USERNAME.getKey(), "root");
+    jdbcProps.put(JdbcConfig.PASSWORD.getKey(), "");
+
+    CatalogEntity jdbcEntity =
+        CatalogEntity.builder()
+            .withId(7L)
+            .withName("jdbc-catalog-empty-password")
+            .withNamespace(Namespace.of("metalake"))
+            .withType(TestableJdbcCatalog.Type.RELATIONAL)
+            .withProvider("jdbc-starrocks")
+            .withAuditInfo(auditInfo)
+            .withProperties(jdbcProps)
+            .build();
+
+    TestableJdbcCatalog jdbcCatalog = new TestableJdbcCatalog();
+    jdbcCatalog.withCatalogConf(jdbcProps).withCatalogEntity(jdbcEntity);
+    Map<String, String> properties = jdbcCatalog.propertiesWithCredentialProviders();
+
+    String credentialProviders = properties.get(CredentialConstants.CREDENTIAL_PROVIDERS);
+    Assertions.assertNotNull(credentialProviders);
+    Assertions.assertEquals(JdbcCredential.JDBC_CREDENTIAL_TYPE, credentialProviders);
+  }
+
   @Test
   void testJdbcCatalogNoCredentialProvidersWithoutPassword() {
     AuditInfo auditInfo =

catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalog.java

@@ -18,8 +18,6 @@
  */
 package org.apache.gravitino.catalog.lakehouse.iceberg;
 
-import com.google.common.collect.Maps;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import org.apache.commons.lang3.StringUtils;
@@ -93,40 +91,25 @@ public PropertiesMetadata schemaPropertiesMetadata() throws UnsupportedOperation
   @Override
   @Evolving
   public Map<String, String> propertiesWithCredentialProviders() {
-    Map<String, String> properties = Maps.newHashMap(super.propertiesWithCredentialProviders());
-    // Iceberg is security-first: the vended s3:ListBucket statement keeps the bare location prefix
-    // disabled so a credential cannot enumerate sibling keys sharing the location prefix. This is
-    // determined by the catalog type and is not meant to be configured by users.
-    properties.put(CredentialConstants.S3_CREDENTIAL_LIST_LOCATION_PREFIX, "false");
-    return applyDefaultCredentialProviders(properties);
+    Map<String, String> props = super.propertiesWithCredentialProviders();
+    // Iceberg is security-first: disable s3:ListBucket on bare location prefix so a vended
+    // credential cannot enumerate sibling keys. This is catalog-type policy, not user-configurable.
+    props.put(CredentialConstants.S3_CREDENTIAL_LIST_LOCATION_PREFIX, "false");
+    return props;
   }
 
-  private Map<String, String> applyDefaultCredentialProviders(Map<String, String> properties) {
-    // If credential providers already set, return as is
-    if (StringUtils.isNotBlank(properties.get(CredentialConstants.CREDENTIAL_PROVIDERS))) {
-      return properties;
-    }
-
-    List<String> credentialProviders = new ArrayList<>();
-
-    // Add JDBC credential provider if backend is JDBC and jdbc-user/jdbc-password are set
+  @Override
+  protected void addCatalogSpecificCredentialProviders(
+      Map<String, String> properties, List<String> credentialProviders) {
     String catalogBackend = properties.get(IcebergConstants.CATALOG_BACKEND);
     if (catalogBackend != null
         && IcebergCatalogBackend.JDBC.name().equalsIgnoreCase(catalogBackend)) {
       String jdbcUser = properties.get(IcebergConstants.GRAVITINO_JDBC_USER);
       String jdbcPassword = properties.get(IcebergConstants.GRAVITINO_JDBC_PASSWORD);
-      if (StringUtils.isNotBlank(jdbcUser) && StringUtils.isNotBlank(jdbcPassword)) {
+      if (StringUtils.isNotBlank(jdbcUser) && jdbcPassword != null) {
         credentialProviders.add(JdbcCredential.JDBC_CREDENTIAL_TYPE);
       }
     }
-
     addStorageCredentialProviders(properties, credentialProviders);
-
-    if (!credentialProviders.isEmpty()) {
-      properties.put(
-          CredentialConstants.CREDENTIAL_PROVIDERS, String.join(",", credentialProviders));
-    }
-
-    return properties;
   }
 }

catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogPropertiesMetadata.java

@@ -29,14 +29,15 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import org.apache.gravitino.cloud.storage.AzurePropertiesMetadata;
+import org.apache.gravitino.cloud.storage.GCSPropertiesMetadata;
+import org.apache.gravitino.cloud.storage.OSSPropertiesMetadata;
+import org.apache.gravitino.cloud.storage.S3PropertiesMetadata;
 import org.apache.gravitino.connector.BaseCatalogPropertiesMetadata;
 import org.apache.gravitino.connector.PropertyEntry;
 import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig;
 import org.apache.gravitino.iceberg.common.authentication.kerberos.KerberosConfig;
 import org.apache.gravitino.iceberg.common.cache.LocalTableMetadataCache;
-import org.apache.gravitino.storage.AzureProperties;
-import org.apache.gravitino.storage.OSSProperties;
-import org.apache.gravitino.storage.S3Properties;
 
 public class IcebergCatalogPropertiesMetadata extends BaseCatalogPropertiesMetadata {
   public static final String CATALOG_BACKEND = IcebergConstants.CATALOG_BACKEND;
@@ -89,41 +90,17 @@ public class IcebergCatalogPropertiesMetadata extends BaseCatalogPropertiesMetad
                 null /* defaultValue */,
                 false /* hidden */),
             stringOptionalPropertyEntry(
-                S3Properties.GRAVITINO_S3_ACCESS_KEY_ID,
-                "s3 access key ID",
+                GRAVITINO_JDBC_USER,
+                "JDBC user for Iceberg JDBC backend",
                 false /* immutable */,
                 null /* defaultValue */,
-                false /* hidden */),
-            stringOptionalPropertyEntry(
-                S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY,
-                "s3 secret access key",
-                false /* immutable */,
-                null /* defaultValue */,
-                false /* hidden */),
-            stringOptionalPropertyEntry(
-                OSSProperties.GRAVITINO_OSS_ACCESS_KEY_ID,
-                "OSS access key ID",
-                false /* immutable */,
-                null /* defaultValue */,
-                false /* hidden */),
+                true /* hidden */),
             stringOptionalPropertyEntry(
-                OSSProperties.GRAVITINO_OSS_ACCESS_KEY_SECRET,
-                "OSS access key secret",
+                GRAVITINO_JDBC_PASSWORD,
+                "JDBC password for Iceberg JDBC backend",
                 false /* immutable */,
                 null /* defaultValue */,
-                false /* hidden */),
-            stringOptionalPropertyEntry(
-                AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_NAME,
-                "Azure storage account name",
-                false /* immutable */,
-                null /* defaultValue */,
-                false /* hidden */),
-            stringOptionalPropertyEntry(
-                AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_KEY,
-                "Azure storage account key",
-                false /* immutable */,
-                null /* defaultValue */,
-                false /* hidden */),
+                true /* hidden */),
             stringOptionalPropertyEntry(
                 IcebergConstants.TABLE_METADATA_CACHE_IMPL,
                 "Table metadata cache implementation. Set to empty string(\"\") if "
@@ -153,6 +130,10 @@ public class IcebergCatalogPropertiesMetadata extends BaseCatalogPropertiesMetad
                 false));
     HashMap<String, PropertyEntry<?>> result = Maps.newHashMap();
     result.putAll(Maps.uniqueIndex(propertyEntries, PropertyEntry::getName));
+    result.putAll(S3PropertiesMetadata.PROPERTY_ENTRIES);
+    result.putAll(OSSPropertiesMetadata.PROPERTY_ENTRIES);
+    result.putAll(AzurePropertiesMetadata.PROPERTY_ENTRIES);
+    result.putAll(GCSPropertiesMetadata.PROPERTY_ENTRIES);
     result.putAll(KerberosConfig.KERBEROS_PROPERTY_ENTRIES);
     result.putAll(AuthenticationConfig.AUTHENTICATION_PROPERTY_ENTRIES);
     PROPERTIES_METADATA = ImmutableMap.copyOf(result);