Reject reserved Hadoop keys in GCS fileSystemProperties

src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java

@@ -38,9 +38,12 @@
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import com.google.common.io.ByteStreams;
 import com.google.gson.reflect.TypeToken;
+import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.plugin.gcp.gcs.GCSPath;
 import io.cdap.plugin.gcp.gcs.ServiceAccountAccessTokenProvider;
 import org.apache.hadoop.conf.Configuration;
@@ -109,6 +112,47 @@ public class GCPUtils {
   private static final Pattern IAM_IMPERSONATION_URL_PATTERN = Pattern.compile(
     "^https://iamcredentials(\\.mtls)?\\.googleapis\\.com/v1/projects/-/serviceAccounts/[^/?#:]+:generateAccessToken$");
 
+  /**
+   * Hadoop configuration keys that users are not allowed to set through the 'fileSystemProperties' plugin
+   * property. These keys control where the GCS connector sends requests and the bearer token that authenticates
+   * them, or which credentials and classes it loads. Letting a user override them would allow the credentials of
+   * the environment the plugin runs in (for example, the service account of the CDAP system service during
+   * pipeline validation) to be sent to an arbitrary endpoint or used to impersonate another identity.
+   */
+  private static final Set<String> RESERVED_FS_PROPERTY_KEYS = ImmutableSet.of(
+    // GCS API endpoint. The connector sends 'Authorization: Bearer <token>' to whatever is configured here.
+    "fs.gs.storage.root.url",
+    "fs.gs.storage.service.path",
+    // OAuth token endpoint used to obtain or refresh access tokens.
+    "fs.gs.token.server.url",
+    // HTTP transport used by the credential and proxy code paths.
+    "fs.gs.http.transport.type",
+    // Deprecated aliases of 'fs.gs.auth.service.account.*' keys that the connector still honors.
+    "fs.gs.enable.service.account.auth",
+    "fs.gs.service.account.auth.email",
+    "fs.gs.service.account.auth.keyfile",
+    // FileSystem implementation classes.
+    "fs.gs.impl",
+    "fs.AbstractFileSystem.gs.impl");
+
+  /**
+   * Prefixes of Hadoop configuration keys that users are not allowed to set through the 'fileSystemProperties'
+   * plugin property. See {@link #RESERVED_FS_PROPERTY_KEYS}.
+   */
+  private static final List<String> RESERVED_FS_PROPERTY_PREFIXES = ImmutableList.of(
+    // Proxy address, username and password. Routes every connector request through a user-controlled host.
+    "fs.gs.proxy.",
+    // Credential sources, access token provider class and service account impersonation.
+    "fs.gs.auth.",
+    // Classes instantiated by the connector.
+    "fs.gs.authorization.handler.",
+    "fs.gs.delegation.token.",
+    // The connector resolves every 'fs.gs.<suffix>' credential key through 'google.cloud.<suffix>' as well,
+    // so the fallback prefix has to be reserved too or the keys above could be set through it.
+    "google.cloud.",
+    // Keys this class uses to pass the service account of the plugin to ServiceAccountAccessTokenProvider.
+    "cdap.auth.");
+
   /**
    * Load a service account from the local file system.
    *
@@ -306,6 +350,12 @@ public static Map<String, String> getFileSystemProperties(GCPConnectorConfig con
   private static Map<String, String> getFileSystemProperties(String project, String serviceAccount,
                                                              String serviceAccountType, String path,
                                                              Map<String, String> properties) {
+    // The given properties can contain user-supplied entries. Reserved keys are rejected by
+    // validateFileSystemProperties() at validation time; dropping them here as well guarantees they never reach
+    // the Hadoop configuration, even for values that were resolved from macros after validation.
+    // Work on a copy so that the caller's map is not mutated and immutable or null maps are accepted.
+    properties = properties == null ? new HashMap<>() : new HashMap<>(properties);
+    properties.keySet().removeIf(GCPUtils::isReservedFileSystemProperty);
     try {
       properties.putAll(generateGCSAuthProperties(serviceAccount, serviceAccountType));
     } catch (Exception ignored) {
@@ -321,6 +371,52 @@ private static Map<String, String> getFileSystemProperties(String project, Strin
     return properties;
   }
 
+  /**
+   * Returns whether the given Hadoop configuration key is reserved and must not be set by users through the
+   * 'fileSystemProperties' plugin property.
+   *
+   * @param key the Hadoop configuration key
+   * @return true if the key is reserved
+   */
+  public static boolean isReservedFileSystemProperty(@Nullable String key) {
+    if (key == null) {
+      return false;
+    }
+    String trimmedKey = key.trim();
+    if (RESERVED_FS_PROPERTY_KEYS.contains(trimmedKey)) {
+      return true;
+    }
+    for (String prefix : RESERVED_FS_PROPERTY_PREFIXES) {
+      if (trimmedKey.startsWith(prefix)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  /**
+   * Validates that user-supplied file system properties do not set any reserved Hadoop configuration key.
+   * One failure is added to the collector for every reserved key found, so that all offending keys are reported
+   * at once.
+   *
+   * @param properties the user-supplied file system properties, may be null
+   * @param configProperty the name of the plugin property the file system properties were read from
+   * @param collector the failure collector to report reserved keys to
+   */
+  public static void validateFileSystemProperties(@Nullable Map<String, String> properties, String configProperty,
+                                                  FailureCollector collector) {
+    if (properties == null) {
+      return;
+    }
+    for (String key : properties.keySet()) {
+      if (isReservedFileSystemProperty(key)) {
+        collector.addFailure(
+          String.format("File system property '%s' is reserved and cannot be overridden.", key),
+          "Remove it from the file system properties.").withConfigProperty(configProperty);
+      }
+    }
+  }
+
   public static BigQuery getBigQuery(String project, @Nullable Credentials credentials, @Nullable Integer readTimeout) {
     BigQueryOptions.Builder bigqueryBuilder = BigQueryOptions.newBuilder().setProjectId(project);
     if (credentials != null) {

src/main/java/io/cdap/plugin/gcp/gcs/sink/GCSBatchSink.java

@@ -177,7 +177,16 @@ protected Map<String, String> getFileSystemProperties(BatchSinkContext context)
     Map<String, String> properties = GCPUtils.getFileSystemProperties(config.connection, config.getPath(),
                                                                       new HashMap<>());
     properties.put(GCSBatchSink.CONTENT_TYPE, config.getContentType());
-    properties.putAll(config.getFileSystemProperties());
+    // user-supplied properties are merged last so they can tune the output format, but reserved keys are never
+    // taken from the user. They are rejected during validation; this guards values resolved from macros afterwards.
+    Map<String, String> userProperties = config.getFileSystemProperties();
+    if (userProperties != null) {
+      for (Map.Entry<String, String> entry : userProperties.entrySet()) {
+        if (!GCPUtils.isReservedFileSystemProperty(entry.getKey())) {
+          properties.put(entry.getKey(), entry.getValue());
+        }
+      }
+    }
     String outputFileBaseName = config.getOutputFileNameBase();
     if (outputFileBaseName == null || outputFileBaseName.isEmpty()) {
       return properties;
@@ -456,7 +465,8 @@ public void validate(FailureCollector collector, Map<String, String> arguments)
       }
 
       try {
-        getFileSystemProperties();
+        Map<String, String> fileSystemProperties = getFileSystemProperties();
+        GCPUtils.validateFileSystemProperties(fileSystemProperties, NAME_FS_PROPERTIES, collector);
       } catch (IllegalArgumentException e) {
         collector.addFailure("File system properties must be a valid json.", null)
           .withConfigProperty(NAME_FS_PROPERTIES).withStacktrace(e.getStackTrace());

src/main/java/io/cdap/plugin/gcp/gcs/source/GCSSource.java

@@ -308,7 +308,8 @@ public void validate(FailureCollector collector) {
       }
       if (!containsMacro(NAME_FILE_SYSTEM_PROPERTIES)) {
         try {
-          getFileSystemProperties();
+          Map<String, String> fileSystemProperties = getFileSystemProperties();
+          GCPUtils.validateFileSystemProperties(fileSystemProperties, NAME_FILE_SYSTEM_PROPERTIES, collector);
         } catch (Exception e) {
           collector.addFailure("File system properties must be a valid json.", null)
             .withConfigProperty(NAME_FILE_SYSTEM_PROPERTIES).withStacktrace(e.getStackTrace());

src/test/java/io/cdap/plugin/gcp/common/GCPUtilsFileSystemPropertiesTest.java

@@ -0,0 +1,160 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.gcp.common;
+
+import io.cdap.cdap.etl.api.validation.CauseAttributes;
+import io.cdap.cdap.etl.api.validation.ValidationFailure;
+import io.cdap.cdap.etl.mock.validation.MockFailureCollector;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Tests for the handling of reserved keys in user-supplied file system properties.
+ */
+public class GCPUtilsFileSystemPropertiesTest {
+
+  private static final String CONFIG_PROPERTY = "fileSystemProperties";
+
+  private static final List<String> RESERVED_KEYS = Arrays.asList(
+    "fs.gs.storage.root.url",
+    "fs.gs.storage.service.path",
+    "fs.gs.token.server.url",
+    "fs.gs.http.transport.type",
+    "fs.gs.enable.service.account.auth",
+    "fs.gs.service.account.auth.email",
+    "fs.gs.service.account.auth.keyfile",
+    "fs.gs.impl",
+    "fs.AbstractFileSystem.gs.impl",
+    "fs.gs.proxy.address",
+    "fs.gs.proxy.username",
+    "fs.gs.proxy.password",
+    "fs.gs.auth.access.token.provider.impl",
+    "fs.gs.auth.impersonation.service.account",
+    "fs.gs.auth.service.account.json.keyfile",
+    "fs.gs.auth.service.account.private.key",
+    "fs.gs.auth.client.secret",
+    "fs.gs.auth.refresh.token",
+    "fs.gs.authorization.handler.impl",
+    "fs.gs.delegation.token.binding",
+    "google.cloud.token.server.url",
+    "google.cloud.auth.access.token.provider.impl",
+    "google.cloud.proxy.address",
+    "cdap.auth.service.account",
+    "cdap.auth.service.account.type",
+    "cdap.auth.service.account.scopes");
+
+  private static final List<String> ALLOWED_KEYS = Arrays.asList(
+    "fs.gs.inputstream.fadvise",
+    "fs.gs.outputstream.upload.chunk.size",
+    "fs.gs.block.size",
+    "fs.gs.http.max.retry",
+    "fs.gs.http.connect-timeout",
+    "fs.gs.status.parallel.enable",
+    "fs.gs.project.id",
+    "fs.gs.system.bucket",
+    "mapreduce.input.fileinputformat.split.minsize",
+    "avro.mo.config.namedOutput");
+
+  @Test
+  public void testReservedKeysAreDetected() {
+    for (String key : RESERVED_KEYS) {
+      Assert.assertTrue(key + " should be reserved", GCPUtils.isReservedFileSystemProperty(key));
+    }
+  }
+
+  @Test
+  public void testAllowedKeysAreNotDetected() {
+    for (String key : ALLOWED_KEYS) {
+      Assert.assertFalse(key + " should not be reserved", GCPUtils.isReservedFileSystemProperty(key));
+    }
+    Assert.assertFalse(GCPUtils.isReservedFileSystemProperty(null));
+    Assert.assertFalse(GCPUtils.isReservedFileSystemProperty(""));
+  }
+
+  @Test
+  public void testSurroundingWhitespaceDoesNotBypassReservedKeys() {
+    Assert.assertTrue(GCPUtils.isReservedFileSystemProperty(" fs.gs.storage.root.url "));
+    Assert.assertTrue(GCPUtils.isReservedFileSystemProperty("\tfs.gs.token.server.url"));
+  }
+
+  @Test
+  public void testValidateReportsOneFailurePerReservedKey() {
+    Map<String, String> properties = new HashMap<>();
+    properties.put("fs.gs.storage.root.url", "https://attacker.example.com/");
+    properties.put("fs.gs.token.server.url", "https://attacker.example.com/token");
+    properties.put("fs.gs.inputstream.fadvise", "AUTO");
+
+    MockFailureCollector collector = new MockFailureCollector("gcs");
+    GCPUtils.validateFileSystemProperties(properties, CONFIG_PROPERTY, collector);
+
+    List<ValidationFailure> failures = collector.getValidationFailures();
+    Assert.assertEquals(2, failures.size());
+    for (ValidationFailure failure : failures) {
+      Assert.assertEquals(1, failure.getCauses().size());
+      Assert.assertEquals(CONFIG_PROPERTY, failure.getCauses().get(0).getAttribute(CauseAttributes.STAGE_CONFIG));
+      Assert.assertTrue(failure.getMessage().contains("is reserved and cannot be overridden"));
+    }
+  }
+
+  @Test
+  public void testValidateAcceptsAllowedKeys() {
+    Map<String, String> properties = new HashMap<>();
+    for (String key : ALLOWED_KEYS) {
+      properties.put(key, "value");
+    }
+    MockFailureCollector collector = new MockFailureCollector("gcs");
+    GCPUtils.validateFileSystemProperties(properties, CONFIG_PROPERTY, collector);
+    Assert.assertEquals(0, collector.getValidationFailures().size());
+  }
+
+  @Test
+  public void testValidateAcceptsNullAndEmptyProperties() {
+    MockFailureCollector collector = new MockFailureCollector("gcs");
+    GCPUtils.validateFileSystemProperties(null, CONFIG_PROPERTY, collector);
+    GCPUtils.validateFileSystemProperties(new HashMap<>(), CONFIG_PROPERTY, collector);
+    Assert.assertEquals(0, collector.getValidationFailures().size());
+  }
+
+  @Test
+  public void testGetFileSystemPropertiesDropsReservedUserKeys() {
+    Map<String, String> userProperties = new HashMap<>();
+    userProperties.put("fs.gs.storage.root.url", "https://attacker.example.com/");
+    userProperties.put("google.cloud.token.server.url", "https://attacker.example.com/token");
+    userProperties.put("cdap.auth.service.account", "{\"type\":\"external_account\"}");
+    userProperties.put("fs.gs.auth.access.token.provider.impl", "com.example.EvilProvider");
+    userProperties.put("fs.gs.inputstream.fadvise", "AUTO");
+
+    GCPConnectorConfig connection = new GCPConnectorConfig("my-project", "filePath", "auto-detect", null);
+    Map<String, String> properties = GCPUtils.getFileSystemProperties(connection, "gs://bucket/path",
+                                                                      userProperties);
+
+    Assert.assertFalse(properties.containsKey("fs.gs.storage.root.url"));
+    Assert.assertFalse(properties.containsKey("google.cloud.token.server.url"));
+    Assert.assertFalse(properties.containsKey("cdap.auth.service.account"));
+    // the access token provider is always the one set by the plugin
+    Assert.assertEquals("io.cdap.plugin.gcp.gcs.ServiceAccountAccessTokenProvider",
+                        properties.get("fs.gs.auth.access.token.provider.impl"));
+    Assert.assertEquals("AUTO", properties.get("fs.gs.inputstream.fadvise"));
+    Assert.assertEquals("my-project", properties.get("fs.gs.project.id"));
+    Assert.assertEquals("bucket", properties.get("fs.gs.system.bucket"));
+  }
+}

src/test/java/io/cdap/plugin/gcp/gcs/sink/GCSBatchSinkTest.java

@@ -60,6 +60,38 @@ public void testInvalidFSProperties() {
     Assert.assertEquals(1, collector.getValidationFailures().size());
   }
 
+  @Test
+  public void testReservedFSPropertiesAreRejected() {
+    String[] reservedKeys = {
+      "fs.gs.storage.root.url",
+      "fs.gs.token.server.url",
+      "google.cloud.token.server.url",
+      "fs.gs.auth.access.token.provider.impl",
+      "cdap.auth.service.account"
+    };
+    for (String key : reservedKeys) {
+      GCSBatchSinkConfig config = getBuilder("{\"" + key + "\":\"https://evil.example.com/\"}").build();
+      MockFailureCollector collector = new MockFailureCollector("gcssink");
+      config.validate(collector);
+      List<ValidationFailure> failures = collector.getValidationFailures();
+      Assert.assertEquals(key + " should be rejected", 1, failures.size());
+      Assert.assertTrue(failures.get(0).getMessage().contains(key));
+      List<ValidationFailure.Cause> causes = failures.get(0).getCauses();
+      Assert.assertEquals(1, causes.size());
+      Assert.assertEquals("fileSystemProperties", causes.get(0).getAttribute(CauseAttributes.STAGE_CONFIG));
+    }
+  }
+
+  @Test
+  public void testEveryReservedFSPropertyIsReported() {
+    GCSBatchSinkConfig config = getBuilder("{\"fs.gs.storage.root.url\":\"https://evil.example.com/\","
+                                             + "\"fs.gs.proxy.address\":\"evil.example.com:8080\","
+                                             + "\"fs.gs.outputstream.upload.chunk.size\":\"1048576\"}").build();
+    MockFailureCollector collector = new MockFailureCollector("gcssink");
+    config.validate(collector);
+    Assert.assertEquals(2, collector.getValidationFailures().size());
+  }
+
   @Test
   public void testValidContentType() {
     GCSBatchSinkConfig.Builder builder = getBuilder(null);