diff --git a/src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java b/src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java index b94381587f..55bb595d10 100644 --- a/src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java +++ b/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 RESERVED_FS_PROPERTY_KEYS = ImmutableSet.of( + // GCS API endpoint. The connector sends 'Authorization: Bearer ' 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 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.' credential key through 'google.cloud.' 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 getFileSystemProperties(GCPConnectorConfig con private static Map getFileSystemProperties(String project, String serviceAccount, String serviceAccountType, String path, Map 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 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 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) { diff --git a/src/main/java/io/cdap/plugin/gcp/gcs/sink/GCSBatchSink.java b/src/main/java/io/cdap/plugin/gcp/gcs/sink/GCSBatchSink.java index 96b555e4a4..d250372fb5 100644 --- a/src/main/java/io/cdap/plugin/gcp/gcs/sink/GCSBatchSink.java +++ b/src/main/java/io/cdap/plugin/gcp/gcs/sink/GCSBatchSink.java @@ -177,7 +177,16 @@ protected Map getFileSystemProperties(BatchSinkContext context) Map 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 userProperties = config.getFileSystemProperties(); + if (userProperties != null) { + for (Map.Entry 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 arguments) } try { - getFileSystemProperties(); + Map 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()); diff --git a/src/main/java/io/cdap/plugin/gcp/gcs/source/GCSSource.java b/src/main/java/io/cdap/plugin/gcp/gcs/source/GCSSource.java index 50142466a9..a548d5e06c 100644 --- a/src/main/java/io/cdap/plugin/gcp/gcs/source/GCSSource.java +++ b/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 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()); diff --git a/src/test/java/io/cdap/plugin/gcp/common/GCPUtilsFileSystemPropertiesTest.java b/src/test/java/io/cdap/plugin/gcp/common/GCPUtilsFileSystemPropertiesTest.java new file mode 100644 index 0000000000..14336beef7 --- /dev/null +++ b/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 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 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 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 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 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 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 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")); + } +} diff --git a/src/test/java/io/cdap/plugin/gcp/gcs/sink/GCSBatchSinkTest.java b/src/test/java/io/cdap/plugin/gcp/gcs/sink/GCSBatchSinkTest.java index e0377f5a31..cba5c3b2cf 100644 --- a/src/test/java/io/cdap/plugin/gcp/gcs/sink/GCSBatchSinkTest.java +++ b/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 failures = collector.getValidationFailures(); + Assert.assertEquals(key + " should be rejected", 1, failures.size()); + Assert.assertTrue(failures.get(0).getMessage().contains(key)); + List 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); diff --git a/src/test/java/io/cdap/plugin/gcp/gcs/source/GCSSourceConfigTest.java b/src/test/java/io/cdap/plugin/gcp/gcs/source/GCSSourceConfigTest.java new file mode 100644 index 0000000000..f75040137d --- /dev/null +++ b/src/test/java/io/cdap/plugin/gcp/gcs/source/GCSSourceConfigTest.java @@ -0,0 +1,127 @@ +/* + * 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.gcs.source; + +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 io.cdap.plugin.format.plugin.AbstractFileSourceConfig; +import io.cdap.plugin.gcp.gcs.source.GCSSource.GCSSourceConfig; +import org.junit.Assert; +import org.junit.Test; + +import java.lang.reflect.Field; +import java.util.List; +import javax.annotation.Nullable; + +/** + * Tests for {@link GCSSourceConfig} validation. + */ +public class GCSSourceConfigTest { + + private static final String FS_PROPERTIES = "fileSystemProperties"; + + private GCSSourceConfig getConfig(@Nullable String fileSystemProperties) throws Exception { + GCSSourceConfig config = new GCSSourceConfig(); + setField(config, AbstractFileSourceConfig.class, "referenceName", "testref"); + setField(config, GCSSourceConfig.class, "path", "gs://test/path"); + setField(config, GCSSourceConfig.class, "fileSystemProperties", fileSystemProperties); + return config; + } + + private static void setField(Object target, Class clz, String name, @Nullable Object value) throws Exception { + Field field = clz.getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } + + @Test + public void testNoFSProperties() throws Exception { + MockFailureCollector collector = new MockFailureCollector("gcssource"); + getConfig(null).validate(collector); + Assert.assertEquals(0, collector.getValidationFailures().size()); + } + + @Test + public void testAllowedFSProperties() throws Exception { + MockFailureCollector collector = new MockFailureCollector("gcssource"); + getConfig("{\"fs.gs.inputstream.fadvise\":\"AUTO\",\"mapreduce.input.fileinputformat.split.minsize\":\"1\"}") + .validate(collector); + Assert.assertEquals(0, collector.getValidationFailures().size()); + } + + @Test + public void testInvalidJsonFSProperties() throws Exception { + MockFailureCollector collector = new MockFailureCollector("gcssource"); + getConfig("{\"key\":}").validate(collector); + Assert.assertEquals(1, collector.getValidationFailures().size()); + assertFailureOnFSProperties(collector.getValidationFailures().get(0)); + } + + @Test + public void testReservedStorageRootUrlIsRejected() throws Exception { + assertSingleReservedKeyFailure("fs.gs.storage.root.url"); + } + + @Test + public void testReservedTokenServerUrlIsRejected() throws Exception { + assertSingleReservedKeyFailure("fs.gs.token.server.url"); + } + + @Test + public void testReservedKeyThroughFallbackPrefixIsRejected() throws Exception { + assertSingleReservedKeyFailure("google.cloud.token.server.url"); + } + + @Test + public void testReservedPluginAuthKeyIsRejected() throws Exception { + assertSingleReservedKeyFailure("cdap.auth.service.account"); + } + + @Test + public void testReservedImpersonationKeyIsRejected() throws Exception { + assertSingleReservedKeyFailure("fs.gs.auth.impersonation.service.account"); + } + + @Test + public void testEveryReservedKeyIsReported() throws Exception { + MockFailureCollector collector = new MockFailureCollector("gcssource"); + getConfig("{\"fs.gs.storage.root.url\":\"https://evil.example.com/\"," + + "\"fs.gs.proxy.address\":\"evil.example.com:8080\"," + + "\"fs.gs.inputstream.fadvise\":\"AUTO\"}").validate(collector); + List failures = collector.getValidationFailures(); + Assert.assertEquals(2, failures.size()); + for (ValidationFailure failure : failures) { + assertFailureOnFSProperties(failure); + } + } + + private void assertSingleReservedKeyFailure(String key) throws Exception { + MockFailureCollector collector = new MockFailureCollector("gcssource"); + getConfig("{\"" + key + "\":\"https://evil.example.com/\"}").validate(collector); + List failures = collector.getValidationFailures(); + Assert.assertEquals(1, failures.size()); + assertFailureOnFSProperties(failures.get(0)); + Assert.assertTrue(failures.get(0).getMessage().contains(key)); + } + + private void assertFailureOnFSProperties(ValidationFailure failure) { + List causes = failure.getCauses(); + Assert.assertEquals(1, causes.size()); + Assert.assertEquals(FS_PROPERTIES, causes.get(0).getAttribute(CauseAttributes.STAGE_CONFIG)); + } +}