[#11642] improvement(fileset): Hide credential properties from fileset catalog properties

bundles/aliyun/src/main/java/org/apache/gravitino/oss/fs/OSSFileSystemProvider.java

@@ -87,6 +87,14 @@ public Map<String, String> getFileSystemCredentialConf(Credential[] credentials)
     return result;
   }
 
+  @Override
+  public boolean containsClientCredentials(Map<String, String> config) {
+    return SupportsCredentialVending.allNonBlank(
+        config,
+        OSSProperties.GRAVITINO_OSS_ACCESS_KEY_ID,
+        OSSProperties.GRAVITINO_OSS_ACCESS_KEY_SECRET);
+  }
+
   @Override
   public String scheme() {
     return "oss";

bundles/aliyun/src/test/java/org/apache/gravitino/oss/fs/TestOSSFileSystemProvider.java

@@ -0,0 +1,48 @@
+/*
+ * 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.gravitino.oss.fs;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.gravitino.storage.OSSProperties;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestOSSFileSystemProvider {
+
+  private final OSSFileSystemProvider provider = new OSSFileSystemProvider();
+
+  @Test
+  public void testContainsClientCredentialsTrueWhenBothKeysPresent() {
+    Assertions.assertTrue(
+        provider.containsClientCredentials(
+            ImmutableMap.of(
+                OSSProperties.GRAVITINO_OSS_ACCESS_KEY_ID, "ak",
+                OSSProperties.GRAVITINO_OSS_ACCESS_KEY_SECRET, "sk")));
+  }
+
+  @Test
+  public void testContainsClientCredentialsFalseWhenIncomplete() {
+    Assertions.assertFalse(
+        provider.containsClientCredentials(
+            ImmutableMap.of(OSSProperties.GRAVITINO_OSS_ACCESS_KEY_ID, "ak")));
+    Assertions.assertFalse(
+        provider.containsClientCredentials(
+            ImmutableMap.of(OSSProperties.GRAVITINO_OSS_ENDPOINT, "endpoint")));
+  }
+}

bundles/aws/src/main/java/org/apache/gravitino/s3/fs/S3FileSystemProvider.java

@@ -101,6 +101,14 @@ public Map<String, String> getFileSystemCredentialConf(Credential[] credentials)
     return result;
   }
 
+  @Override
+  public boolean containsClientCredentials(Map<String, String> config) {
+    return SupportsCredentialVending.allNonBlank(
+        config,
+        S3Properties.GRAVITINO_S3_ACCESS_KEY_ID,
+        S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY);
+  }
+
   private void checkAndSetCredentialProvider(Map<String, String> configs) {
     String provides = configs.get(S3_CREDENTIAL_KEY);
     if (provides == null) {

bundles/aws/src/test/java/org/apache/gravitino/s3/fs/TestS3FileSystemProvider.java

@@ -0,0 +1,48 @@
+/*
+ * 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.gravitino.s3.fs;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.gravitino.storage.S3Properties;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestS3FileSystemProvider {
+
+  private final S3FileSystemProvider provider = new S3FileSystemProvider();
+
+  @Test
+  public void testContainsClientCredentialsTrueWhenBothKeysPresent() {
+    Assertions.assertTrue(
+        provider.containsClientCredentials(
+            ImmutableMap.of(
+                S3Properties.GRAVITINO_S3_ACCESS_KEY_ID, "ak",
+                S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY, "sk")));
+  }
+
+  @Test
+  public void testContainsClientCredentialsFalseWhenIncomplete() {
+    Assertions.assertFalse(
+        provider.containsClientCredentials(
+            ImmutableMap.of(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID, "ak")));
+    Assertions.assertFalse(
+        provider.containsClientCredentials(
+            ImmutableMap.of(S3Properties.GRAVITINO_S3_ENDPOINT, "endpoint")));
+  }
+}

bundles/azure/src/main/java/org/apache/gravitino/abs/fs/AzureFileSystemProvider.java

@@ -115,6 +115,14 @@ public String name() {
     return ABS_PROVIDER_NAME;
   }
 
+  @Override
+  public boolean containsClientCredentials(Map<String, String> config) {
+    return SupportsCredentialVending.allNonBlank(
+        config,
+        AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_NAME,
+        AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_KEY);
+  }
+
   /**
    * Add additional Azure specific configuration to tune the performance.
    *

bundles/azure/src/test/java/org/apache/gravitino/abs/fs/TestAzureFileSystemProvider.java

@@ -0,0 +1,45 @@
+/*
+ * 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.gravitino.abs.fs;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.gravitino.storage.AzureProperties;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestAzureFileSystemProvider {
+
+  private final AzureFileSystemProvider provider = new AzureFileSystemProvider();
+
+  @Test
+  public void testContainsClientCredentialsTrueWhenBothKeysPresent() {
+    Assertions.assertTrue(
+        provider.containsClientCredentials(
+            ImmutableMap.of(
+                AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_NAME, "name",
+                AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_KEY, "key")));
+  }
+
+  @Test
+  public void testContainsClientCredentialsFalseWhenIncomplete() {
+    Assertions.assertFalse(
+        provider.containsClientCredentials(
+            ImmutableMap.of(AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_NAME, "name")));
+  }
+}

bundles/gcp/src/main/java/org/apache/gravitino/gcs/fs/GCSFileSystemProvider.java

@@ -69,6 +69,12 @@ public Map<String, String> getFileSystemCredentialConf(Credential[] credentials)
     return result;
   }
 
+  @Override
+  public boolean containsClientCredentials(Map<String, String> config) {
+    return SupportsCredentialVending.allNonBlank(
+        config, GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE);
+  }
+
   @Override
   public String scheme() {
     return "gs";

bundles/gcp/src/test/java/org/apache/gravitino/gcs/fs/TestGCSFileSystemProvider.java

@@ -0,0 +1,41 @@
+/*
+ * 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.gravitino.gcs.fs;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.gravitino.storage.GCSProperties;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestGCSFileSystemProvider {
+
+  private final GCSFileSystemProvider provider = new GCSFileSystemProvider();
+
+  @Test
+  public void testContainsClientCredentialsTrueWhenServiceAccountFilePresent() {
+    Assertions.assertTrue(
+        provider.containsClientCredentials(
+            ImmutableMap.of(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/path/to/file")));
+  }
+
+  @Test
+  public void testContainsClientCredentialsFalseWhenAbsent() {
+    Assertions.assertFalse(provider.containsClientCredentials(ImmutableMap.of("other", "value")));
+  }
+}

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

@@ -33,6 +33,10 @@
 import java.util.Map;
 import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
 import org.apache.gravitino.catalog.hadoop.fs.LocalFileSystemProvider;
+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.credential.config.CredentialConfig;
@@ -224,6 +228,10 @@ public class FilesetCatalogPropertiesMetadata extends BaseCatalogPropertiesMetad
           .putAll(KERBEROS_PROPERTY_ENTRIES)
           .putAll(AUTHENTICATION_PROPERTY_ENTRIES)
           .putAll(CredentialConfig.CREDENTIAL_PROPERTY_ENTRIES)
+          .putAll(S3PropertiesMetadata.PROPERTY_ENTRIES)
+          .putAll(OSSPropertiesMetadata.PROPERTY_ENTRIES)
+          .putAll(AzurePropertiesMetadata.PROPERTY_ENTRIES)
+          .putAll(GCSPropertiesMetadata.PROPERTY_ENTRIES)
           .build();
 
   @Override

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

@@ -28,6 +28,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.BasePropertiesMetadata;
 import org.apache.gravitino.connector.PropertyEntry;
 import org.apache.gravitino.credential.config.CredentialConfig;
@@ -75,7 +79,11 @@ protected Map<String, PropertyEntry<?>> specificPropertyEntries() {
                 false /* hidden */))
         .putAll(KERBEROS_PROPERTY_ENTRIES)
         .putAll(AUTHENTICATION_PROPERTY_ENTRIES)
-        .putAll(CredentialConfig.CREDENTIAL_PROPERTY_ENTRIES);
+        .putAll(CredentialConfig.CREDENTIAL_PROPERTY_ENTRIES)
+        .putAll(S3PropertiesMetadata.PROPERTY_ENTRIES)
+        .putAll(OSSPropertiesMetadata.PROPERTY_ENTRIES)
+        .putAll(AzurePropertiesMetadata.PROPERTY_ENTRIES)
+        .putAll(GCSPropertiesMetadata.PROPERTY_ENTRIES);
     return builder.build();
   }
 }

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

@@ -25,6 +25,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.BasePropertiesMetadata;
 import org.apache.gravitino.connector.PropertyEntry;
 import org.apache.gravitino.credential.config.CredentialConfig;
@@ -69,6 +73,10 @@ public class FilesetSchemaPropertiesMetadata extends BasePropertiesMetadata {
           .putAll(KERBEROS_PROPERTY_ENTRIES)
           .putAll(AUTHENTICATION_PROPERTY_ENTRIES)
           .putAll(CredentialConfig.CREDENTIAL_PROPERTY_ENTRIES)
+          .putAll(S3PropertiesMetadata.PROPERTY_ENTRIES)
+          .putAll(OSSPropertiesMetadata.PROPERTY_ENTRIES)
+          .putAll(AzurePropertiesMetadata.PROPERTY_ENTRIES)
+          .putAll(GCSPropertiesMetadata.PROPERTY_ENTRIES)
           .build();
 
   @Override

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

@@ -20,19 +20,24 @@
 
 import com.google.common.collect.Maps;
 import java.time.Instant;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.Catalog;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.credential.CredentialConstants;
+import org.apache.gravitino.credential.S3SecretKeyCredential;
 import org.apache.gravitino.meta.AuditInfo;
 import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.storage.S3Properties;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 /**
- * Verifies that the fileset catalog enables the internal S3 location-prefix flag so a
- * directory-root {@code getFileStatus} HEAD returns 404 instead of 403, regardless of any
- * user-provided value.
+ * Verifies the fileset catalog credential behavior: the internal S3 location-prefix flag is enabled
+ * regardless of any user-provided value, static storage credentials are hidden from the
+ * outward-facing {@code properties()}, and they remain available for credential vending via {@code
+ * propertiesWithCredentialProviders()}.
  */
 public class TestFilesetCatalogCredential {
 
@@ -74,4 +79,39 @@ void testLocationPrefixNotUserConfigurable() {
             .propertiesWithCredentialProviders()
             .get(CredentialConstants.S3_CREDENTIAL_LIST_LOCATION_PREFIX));
   }
+
+  @Test
+  void testStaticCredentialsHiddenFromCatalogProperties() {
+    Map<String, String> properties = Maps.newHashMap();
+    properties.put(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID, "ak");
+    properties.put(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY, "sk");
+    properties.put(S3Properties.GRAVITINO_S3_ENDPOINT, "https://s3.example.com");
+    FilesetCatalogImpl catalog = newCatalog(properties);
+
+    Map<String, String> exposed = catalog.properties();
+    // Sensitive credentials are filtered out from the outward-facing properties.
+    Assertions.assertFalse(exposed.containsKey(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID));
+    Assertions.assertFalse(exposed.containsKey(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY));
+    // Non-sensitive connection info remains visible.
+    Assertions.assertEquals(
+        "https://s3.example.com", exposed.get(S3Properties.GRAVITINO_S3_ENDPOINT));
+  }
+
+  @Test
+  void testStaticCredentialsStillAvailableForCredentialVending() {
+    Map<String, String> properties = Maps.newHashMap();
+    properties.put(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID, "ak");
+    properties.put(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY, "sk");
+    FilesetCatalogImpl catalog = newCatalog(properties);
+
+    // Hiding the credentials does not break credential vending: the raw credentials are still
+    // available to the server-side credential manager, and the matching credential provider is
+    // auto-injected so clients (e.g. GVFS) can obtain vended credentials.
+    Map<String, String> credProps = catalog.propertiesWithCredentialProviders();
+    Assertions.assertEquals("ak", credProps.get(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID));
+    Assertions.assertEquals("sk", credProps.get(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY));
+    List<String> providers =
+        Arrays.asList(credProps.get(CredentialConstants.CREDENTIAL_PROVIDERS).split(","));
+    Assertions.assertTrue(providers.contains(S3SecretKeyCredential.S3_SECRET_KEY_CREDENTIAL_TYPE));
+  }
 }