From a8bbd22d3791c5d0b696f8c0ef5cedb4e82f01b5 Mon Sep 17 00:00:00 2001 From: Dmitry Potepalov Date: Wed, 9 Feb 2022 11:37:27 +0100 Subject: [PATCH] Fix ignored streaming encryption settings in sstableloader Patch by Dmitry Potepalov; reviewed by bereng and brandonwilliams for CASSANDRA-17367 Encryption settings were silently lost in BulkLoadConnectionFactory. Forced the bulk loader test to do some actual streaming besides just handshaking on the native port, the test does not pass without the change in BulkLoadConnectionFactory. --- .build/build-rat.xml | 1 + CHANGES.txt | 1 + build.xml | 1 + .../tools/BulkLoadConnectionFactory.java | 2 +- test/conf/sstableloader_with_encryption.yaml | 7 +++ .../SSTableLoaderEncryptionOptionsTest.java | 60 ++++++++++++++++++- 6 files changed, 68 insertions(+), 4 deletions(-) create mode 100644 test/conf/sstableloader_with_encryption.yaml diff --git a/.build/build-rat.xml b/.build/build-rat.xml index c9880432b1..13c4169abb 100644 --- a/.build/build-rat.xml +++ b/.build/build-rat.xml @@ -62,6 +62,7 @@ + diff --git a/CHANGES.txt b/CHANGES.txt index 97ce6c85ac..71cf70cb61 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -95,6 +95,7 @@ * GossiperTest.testHasVersion3Nodes didn't take into account trunk version changes, fixed to rely on latest version (CASSANDRA-16651) * Update JNA library to 5.9.0 and snappy-java to version 1.1.8.4 (CASSANDRA-17040) Merged from 4.0: + * Fix ignored streaming encryption settings in sstableloader (CASSANDRA-17367) * Streaming tasks handle empty SSTables correctly (CASSANDRA-16349) * Prevent SSTableLoader from doing unnecessary work (CASSANDRA-16349) Merged from 3.0: diff --git a/build.xml b/build.xml index 01e20612bf..2cfb124cf7 100644 --- a/build.xml +++ b/build.xml @@ -1518,6 +1518,7 @@ + diff --git a/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java index cd77d4f065..b8fde66532 100644 --- a/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java +++ b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java @@ -50,7 +50,7 @@ public class BulkLoadConnectionFactory extends NettyStreamingConnectionFactory // does not know which node is in which dc/rack, connecting to SSL port is always the option. OutboundConnectionSettings template = new OutboundConnectionSettings(getByAddress(to)); if (encryptionOptions != null && encryptionOptions.internode_encryption != EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.none) - template = template.withConnectTo(template.to.withPort(secureStoragePort)); + template = template.withConnectTo(template.to.withPort(secureStoragePort)).withEncryption(encryptionOptions); return connect(template, messagingVersion, kind); } diff --git a/test/conf/sstableloader_with_encryption.yaml b/test/conf/sstableloader_with_encryption.yaml new file mode 100644 index 0000000000..3d0fa546c9 --- /dev/null +++ b/test/conf/sstableloader_with_encryption.yaml @@ -0,0 +1,7 @@ +server_encryption_options: + internode_encryption: all + keystore: test/conf/cassandra_ssl_test.keystore + keystore_password: cassandra + truststore: test/conf/cassandra_ssl_test.truststore + truststore_password: cassandra + protocol: TLSv1.2 diff --git a/test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java b/test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java index 1ddc1fb4d7..0da279e253 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java @@ -19,22 +19,30 @@ package org.apache.cassandra.distributed.test; import java.io.IOException; +import java.io.File; import java.util.Collections; +import java.util.List; import com.google.common.collect.ImmutableMap; +import org.apache.commons.io.FileUtils; import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.tools.BulkLoader; import org.apache.cassandra.tools.ToolRunner; +import org.apache.cassandra.service.StorageService; import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertTrue; +import static org.apache.cassandra.distributed.test.ExecUtil.rethrow; +import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; +import static org.apache.cassandra.distributed.shared.AssertUtils.row; public class SSTableLoaderEncryptionOptionsTest extends AbstractEncryptionOptionsImpl { @@ -72,19 +80,22 @@ public class SSTableLoaderEncryptionOptionsTest extends AbstractEncryptionOption CLUSTER.close(); } @Test - public void bulkLoaderSuccessfullyConnectsOverSsl() throws Throwable + public void bulkLoaderSuccessfullyStreamsOverSsl() throws Throwable { + File sstables_to_upload = prepareSstablesForUpload(); ToolRunner.ToolResult tool = ToolRunner.invokeClass(BulkLoader.class, "--nodes", NODES, "--port", Integer.toString(NATIVE_PORT), - "--storage-port", Integer.toString(STORAGE_PORT), + "--ssl-storage-port", Integer.toString(STORAGE_PORT), "--keystore", validKeyStorePath, "--keystore-password", validKeyStorePassword, "--truststore", validTrustStorePath, "--truststore-password", validTrustStorePassword, - "test/data/legacy-sstables/na/legacy_tables/legacy_na_clust"); + "--conf-path", "test/conf/sstableloader_with_encryption.yaml", + sstables_to_upload.getAbsolutePath()); tool.assertOnCleanExit(); assertTrue(tool.getStdout().contains("Summary statistics")); + assertRows(CLUSTER.get(1).executeInternal("SELECT count(*) FROM ssl_upload_tables.test"), row(42L)); } @Test @@ -103,4 +114,47 @@ public class SSTableLoaderEncryptionOptionsTest extends AbstractEncryptionOption assertNotEquals(0, tool.getExitCode()); assertTrue(tool.getStdout().contains("SSLHandshakeException")); } + + private static File prepareSstablesForUpload() throws IOException + { + generateSstables(); + File sstable_dir = copySstablesFromDataDir("test"); + truncateGeneratedTables(); + return sstable_dir; + } + + private static void generateSstables() throws IOException + { + CLUSTER.schemaChange("CREATE KEYSPACE ssl_upload_tables WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}"); + CLUSTER.schemaChange("CREATE TABLE ssl_upload_tables.test (pk int, val text, PRIMARY KEY (pk))"); + for (int i = 0; i < 42; i++) + { + CLUSTER.get(1).executeInternal(String.format("INSERT INTO ssl_upload_tables.test (pk, val) VALUES (%s, '%s')", + i, Integer.toString(i))); + } + CLUSTER.get(1).runOnInstance(rethrow(() -> StorageService.instance.forceKeyspaceFlush("ssl_upload_tables"))); + } + + private static void truncateGeneratedTables() throws IOException + { + CLUSTER.get(1).executeInternal("TRUNCATE ssl_upload_tables.test"); + } + + private static File copySstablesFromDataDir(String table) throws IOException + { + File cfDir = new File("build/test/cassandra/ssl_upload_tables", table); + cfDir.mkdirs(); + List keyspace_dirs = CLUSTER.get(1).callOnInstance(() -> Keyspace.open("ssl_upload_tables").getColumnFamilyStore(table).getDirectories().getCFDirectories()); + for (File srcDir : keyspace_dirs) + { + for (File file : srcDir.listFiles()) + { + if (file.isFile()) + { + FileUtils.copyFileToDirectory(file, cfDir); + } + } + } + return cfDir; + } }