natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.
This commit is contained in:
Dmitry Potepalov 2022-02-09 11:37:27 +01:00 committed by Brandon Williams
parent 4fa541705c
commit 0e249fd15e
6 changed files with 68 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.build/build-rat.xml
View File

@ -61,6 +61,7 @@
<exclude name="**/test/conf/commitlog_compression_LZ4.yaml"/>
<exclude name="**/test/conf/commitlog_compression_Zstd.yaml"/>
<exclude name="**/test/conf/system_keyspaces_directory.yaml"/>
<exclude name="**/test/conf/sstableloader_with_encryption.yaml"/>
<exclude name="**/test/conf/unit-test-conf/test-native-port.yaml"/>
<exclude name="**/test/data/jmxdump/cassandra-3.0-jmx.yaml"/>
<exclude name="**/test/data/jmxdump/cassandra-3.11-jmx.yaml"/>

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0.4
* 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:

1
build.xml
View File

@ -1405,6 +1405,7 @@
<delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/commitlog"/>
<delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/cdc_raw"/>
<delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/data"/>
<delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/ssl_upload_tables"/>
<delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/system_data"/>
<delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/saved_caches"/>
<delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/hints"/>

2
src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
View File

@ -47,7 +47,7 @@ public class BulkLoadConnectionFactory extends DefaultConnectionFactory implemen
// does not know which node is in which dc/rack, connecting to SSL port is always the option.
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 super.createConnection(template, messagingVersion);
}

7
test/conf/sstableloader_with_encryption.yaml Normal file
View File

@ -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

60
test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java
View File

@ -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<File> 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;
}
}