[#12947] fix(catalog): reject unknown catalog properties to prevent credential leaks

catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogPropertiesMetadata.java

@@ -225,6 +225,16 @@ public class FilesetCatalogPropertiesMetadata extends BaseCatalogPropertiesMetad
                   DEFAULT_DISABLE_FILESYSTEM_OPS,
                   false /* hidden */,
                   false /* reserved */))
+          .put(
+              FS_GRAVITINO_PATH_CONFIG_PREFIX,
+              PropertyEntry.stringOptionalPropertyPrefixEntry(
+                  FS_GRAVITINO_PATH_CONFIG_PREFIX,
+                  "Location-scoped filesystem configs: fs.path.config.<name> and"
+                      + " fs.path.config.<name>.<key>",
+                  false /* immutable */,
+                  null /* default value */,
+                  false /* hidden */,
+                  false /* reserved */))
           // The following two are about authentication.
           .putAll(KERBEROS_PROPERTY_ENTRIES)
           .putAll(AUTHENTICATION_PROPERTY_ENTRIES)

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

@@ -18,8 +18,11 @@
  */
 package org.apache.gravitino.catalog.fileset;
 
+import static org.apache.gravitino.catalog.PropertiesMetadataHelpers.validatePropertyForCreate;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.google.common.collect.ImmutableMap;
@@ -91,4 +94,43 @@ void testApiResponseMasksFilesetCatalogS3Secret() {
         HiddenPropertyMaskUtils.MASKED_VALUE,
         response.get(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY));
   }
+
+  @Test
+  void testAcceptsDeclaredS3CredentialsOnCreate() {
+    FilesetCatalogPropertiesMetadata metadata = new FilesetCatalogPropertiesMetadata();
+    Map<String, String> properties =
+        ImmutableMap.of(
+            FilesetCatalogPropertiesMetadata.LOCATION,
+            "s3a://bucket/path",
+            S3Properties.GRAVITINO_S3_ACCESS_KEY_ID,
+            "AKIATEST",
+            S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY,
+            "secret-value");
+    assertDoesNotThrow(() -> validatePropertyForCreate(metadata, properties));
+  }
+
+  @Test
+  void testRejectsUnknownCatalogProperty() {
+    FilesetCatalogPropertiesMetadata metadata = new FilesetCatalogPropertiesMetadata();
+    Map<String, String> properties = ImmutableMap.of("foo-bar", "value");
+
+    IllegalArgumentException exception =
+        assertThrows(
+            IllegalArgumentException.class, () -> validatePropertyForCreate(metadata, properties));
+    assertTrue(exception.getMessage().contains("Unknown properties"));
+    assertTrue(exception.getMessage().contains("foo-bar"));
+  }
+
+  @Test
+  void testAcceptsFsPathConfigPrefix() {
+    FilesetCatalogPropertiesMetadata metadata = new FilesetCatalogPropertiesMetadata();
+    Map<String, String> properties =
+        ImmutableMap.of(
+            FilesetCatalogPropertiesMetadata.FS_GRAVITINO_PATH_CONFIG_PREFIX + "cluster1",
+            "hdfs://cluster1/",
+            FilesetCatalogPropertiesMetadata.FS_GRAVITINO_PATH_CONFIG_PREFIX
+                + "cluster1.config.resource",
+            "/etc/core-site.xml");
+    assertDoesNotThrow(() -> validatePropertyForCreate(metadata, properties));
+  }
 }

catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestGlueCatalogPropertiesMetadata.java

@@ -18,6 +18,8 @@
  */
 package org.apache.gravitino.catalog.glue;
 
+import static org.apache.gravitino.catalog.PropertiesMetadataHelpers.validatePropertyForAlter;
+import static org.apache.gravitino.catalog.PropertiesMetadataHelpers.validatePropertyForCreate;
 import static org.apache.gravitino.catalog.glue.GlueConstants.AWS_ACCESS_KEY_ID;
 import static org.apache.gravitino.catalog.glue.GlueConstants.AWS_GLUE_CATALOG_ID;
 import static org.apache.gravitino.catalog.glue.GlueConstants.AWS_GLUE_ENDPOINT;
@@ -28,10 +30,17 @@
 import static org.apache.gravitino.catalog.glue.GlueConstants.DEFAULT_TABLE_FORMAT_VALUE;
 import static org.apache.gravitino.catalog.glue.GlueConstants.TABLE_FORMAT_FILTER;
 import static org.apache.gravitino.catalog.glue.GlueConstants.WAREHOUSE;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import com.google.common.collect.ImmutableMap;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.gravitino.connector.HiddenPropertyMaskUtils;
+import org.apache.gravitino.storage.S3Properties;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -75,6 +84,12 @@ void testCredentialsAreOptional() {
     assertFalse(metadata.isRequiredProperty(AWS_SECRET_ACCESS_KEY));
   }
 
+  @Test
+  void testCredentialsAreHidden() {
+    assertTrue(metadata.isHiddenProperty(AWS_ACCESS_KEY_ID));
+    assertTrue(metadata.isHiddenProperty(AWS_SECRET_ACCESS_KEY));
+  }
+
   @Test
   void testEndpointIsOptionalAndNotHidden() {
     assertFalse(metadata.isRequiredProperty(AWS_GLUE_ENDPOINT));
@@ -96,4 +111,63 @@ void testTableFormatFilterDefaultValue() {
         metadata.getDefaultValue(TABLE_FORMAT_FILTER),
         "Default table format filter should be 'all'");
   }
+
+  @Test
+  void testRejectsMistypedS3CredentialProperties() {
+    Map<String, String> props =
+        ImmutableMap.of(
+            AWS_REGION,
+            "us-east-1",
+            WAREHOUSE,
+            "s3://bucket/wh",
+            S3Properties.GRAVITINO_S3_ACCESS_KEY_ID,
+            "AKIATEST",
+            S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY,
+            "secret");
+
+    IllegalArgumentException exception =
+        assertThrows(
+            IllegalArgumentException.class, () -> validatePropertyForCreate(metadata, props));
+    assertTrue(exception.getMessage().contains("Unknown properties"));
+    assertTrue(exception.getMessage().contains(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY));
+  }
+
+  @Test
+  void testAcceptsDeclaredAwsCredentialsAndMasksOnRead() {
+    Map<String, String> props =
+        ImmutableMap.of(
+            AWS_REGION,
+            "us-east-1",
+            WAREHOUSE,
+            "s3://bucket/wh",
+            AWS_ACCESS_KEY_ID,
+            "AKIATEST",
+            AWS_SECRET_ACCESS_KEY,
+            "secret");
+    assertDoesNotThrow(() -> validatePropertyForCreate(metadata, props));
+
+    Map<String, String> masked = HiddenPropertyMaskUtils.maskHiddenProperties(props, metadata);
+    assertEquals(HiddenPropertyMaskUtils.MASKED_VALUE, masked.get(AWS_ACCESS_KEY_ID));
+    assertEquals(HiddenPropertyMaskUtils.MASKED_VALUE, masked.get(AWS_SECRET_ACCESS_KEY));
+  }
+
+  @Test
+  void testAlterRejectsMistypedS3CredentialUpsert() {
+    Map<String, String> upserts =
+        ImmutableMap.of(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY, "secret");
+    IllegalArgumentException exception =
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> validatePropertyForAlter(metadata, upserts, Collections.emptyMap()));
+    assertTrue(exception.getMessage().contains(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY));
+  }
+
+  @Test
+  void testAlterAllowsRemovingMistypedS3Credential() {
+    Map<String, String> deletes =
+        ImmutableMap.of(
+            S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY,
+            S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY);
+    assertDoesNotThrow(() -> validatePropertyForAlter(metadata, Collections.emptyMap(), deletes));
+  }
 }

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

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.apache.gravitino.connector.HiddenPropertyMaskUtils;
@@ -76,6 +77,8 @@ public static void validatePropertyForCreate(
         "Properties or property prefixes are required and must be set: %s",
         absentProperties);
 
+    rejectUnknownProperties(propertiesMetadata, properties.keySet());
+
     // use decode function to validate the property values
     for (Map.Entry<String, String> entry : properties.entrySet()) {
       String key = entry.getKey();
@@ -90,8 +93,19 @@ public static void validatePropertyForAlter(
       PropertiesMetadata propertiesMetadata,
       Map<String, String> upserts,
       Map<String, String> deletes) {
+    if (upserts == null) {
+      upserts = Map.of();
+    }
+    if (deletes == null) {
+      deletes = Map.of();
+    }
+
     HiddenPropertyMaskUtils.validateNoMaskedPlaceholders(upserts);
 
+    // Reject undeclared upserts for closed property sets (catalogs). Deletes of undeclared keys
+    // remain allowed so operators can remove previously persisted mistyped secrets.
+    rejectUnknownProperties(propertiesMetadata, upserts.keySet());
+
     for (Map.Entry<String, String> entry : upserts.entrySet()) {
       if (!propertiesMetadata.containsProperty(entry.getKey())) {
         continue;
@@ -119,4 +133,21 @@ public static void validatePropertyForAlter(
       }
     }
   }
+
+  private static void rejectUnknownProperties(
+      PropertiesMetadata propertiesMetadata, Set<String> propertyKeys) {
+    if (!propertiesMetadata.rejectsUnknownProperties() || propertyKeys == null) {
+      return;
+    }
+    List<String> unknownProperties =
+        propertyKeys.stream()
+            .filter(key -> !propertiesMetadata.containsProperty(key))
+            .sorted()
+            .collect(Collectors.toList());
+    Preconditions.checkArgument(
+        unknownProperties.isEmpty(),
+        "Unknown properties are not allowed: %s. Use properties declared by the catalog"
+            + " provider, or a declared prefix such as gravitino.bypass.",
+        unknownProperties);
+  }
 }

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

@@ -23,6 +23,8 @@
 import static org.apache.gravitino.Catalog.CLOUD_REGION_CODE;
 import static org.apache.gravitino.Catalog.PROPERTY_IN_USE;
 import static org.apache.gravitino.Catalog.PROPERTY_PACKAGE;
+import static org.apache.gravitino.connector.BaseCatalog.CATALOG_BYPASS_PREFIX;
+import static org.apache.gravitino.connector.PropertyEntry.stringOptionalPropertyPrefixEntry;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
@@ -42,6 +44,15 @@ public abstract class BaseCatalogPropertiesMetadata extends BasePropertiesMetada
    */
   public static String PROPERTY_METALAKE_IN_USE = "metalake-in-use";
 
+  /** Prefix for Trino connector passthrough catalog properties. */
+  public static final String TRINO_BYPASS_PREFIX = "trino.bypass.";
+
+  /** Prefix for Flink connector passthrough catalog properties. */
+  public static final String FLINK_BYPASS_PREFIX = "flink.bypass.";
+
+  /** Prefix for Spark connector passthrough catalog properties. */
+  public static final String SPARK_BYPASS_PREFIX = "spark.bypass.";
+
   public static final PropertiesMetadata BASIC_CATALOG_PROPERTIES_METADATA =
       new BaseCatalogPropertiesMetadata() {
         @Override
@@ -99,9 +110,42 @@ protected Map<String, PropertyEntry<?>> specificPropertyEntries() {
                   PROPERTY_METALAKE_IN_USE,
                   "The property indicating the metalake that holds the catalog is in use",
                   true /* default value */,
-                  true /* hidden */)),
+                  true /* hidden */),
+              stringOptionalPropertyPrefixEntry(
+                  CATALOG_BYPASS_PREFIX,
+                  "Pass-through properties forwarded to the underlying catalog implementation",
+                  false /* immutable */,
+                  null /* defaultValue */,
+                  false /* hidden */,
+                  false /* reserved */),
+              stringOptionalPropertyPrefixEntry(
+                  TRINO_BYPASS_PREFIX,
+                  "Pass-through properties forwarded to the Trino connector",
+                  false /* immutable */,
+                  null /* defaultValue */,
+                  false /* hidden */,
+                  false /* reserved */),
+              stringOptionalPropertyPrefixEntry(
+                  FLINK_BYPASS_PREFIX,
+                  "Pass-through properties forwarded to the Flink connector",
+                  false /* immutable */,
+                  null /* defaultValue */,
+                  false /* hidden */,
+                  false /* reserved */),
+              stringOptionalPropertyPrefixEntry(
+                  SPARK_BYPASS_PREFIX,
+                  "Pass-through properties forwarded to the Spark connector",
+                  false /* immutable */,
+                  null /* defaultValue */,
+                  false /* hidden */,
+                  false /* reserved */)),
           PropertyEntry::getName);
 
+  @Override
+  public boolean rejectsUnknownProperties() {
+    return true;
+  }
+
   @Override
   public Map<String, PropertyEntry<?>> propertyEntries() {
     if (propertyEntries == null) {

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

@@ -107,6 +107,19 @@ default boolean containsProperty(String propertyName) {
         || getPropertyPrefixEntry(propertyName).isPresent();
   }
 
+  /**
+   * Whether create and alter upserts must use only declared property names (or declared prefixes).
+   *
+   * <p>Catalog metadata returns {@code true} so mistyped credential keys cannot be persisted and
+   * returned unredacted. Table / schema / other open property models keep the default {@code
+   * false}.
+   *
+   * @return true when undeclared property names are rejected on create and alter upsert
+   */
+  default boolean rejectsUnknownProperties() {
+    return false;
+  }
+
   /**
    * Get the value of the property from the given properties map.
    *