From ba9c2db3920b3f65c284c902cdeade8c91102e22 Mon Sep 17 00:00:00 2001 From: AKASHDEEP KAUR Date: Thu, 15 Oct 2026 21:45:22 +0000 Subject: [PATCH] Reject reserved Hadoop keys in GCS fileSystemProperties The fileSystemProperties field of the GCS source and sink is copied into the Hadoop configuration used by the GCS connector. The connector reads the storage endpoint, token server, proxy, credential source, impersonation target and several implementation classes from that configuration, so a user could point the connector at an arbitrary host or identity, including during pipeline validation where it runs with the identity of the CDAP system service. Reserved keys are now rejected with a validation failure on the fileSystemProperties field, one per offending key. As a safeguard, reserved keys are also dropped where user properties are merged with the properties set by the plugin, so values resolved from macros at runtime cannot reach the connector either. --- .../io/cdap/plugin/gcp/common/GCPUtils.java | 96 +++++++++++ .../plugin/gcp/gcs/sink/GCSBatchSink.java | 14 +- .../cdap/plugin/gcp/gcs/source/GCSSource.java | 3 +- .../GCPUtilsFileSystemPropertiesTest.java | 160 ++++++++++++++++++ .../plugin/gcp/gcs/sink/GCSBatchSinkTest.java | 32 ++++ .../gcp/gcs/source/GCSSourceConfigTest.java | 127 ++++++++++++++ 6 files changed, 429 insertions(+), 3 deletions(-) create mode 100644 src/test/java/io/cdap/plugin/gcp/common/GCPUtilsFileSystemPropertiesTest.java create mode 100644 src/test/java/io/cdap/plugin/gcp/gcs/source/GCSSourceConfigTest.java 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)); + } +}