diff --git a/tools/fqltool/src/org/apache/cassandra/fqltool/ConnectionOptions.java b/tools/fqltool/src/org/apache/cassandra/fqltool/ConnectionOptions.java new file mode 100644 index 000000000000..485339d87b13 --- /dev/null +++ b/tools/fqltool/src/org/apache/cassandra/fqltool/ConnectionOptions.java @@ -0,0 +1,203 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.cassandra.fqltool; + +import javax.net.ssl.SSLContext; + +import com.datastax.driver.core.AuthProvider; +import com.datastax.driver.core.RemoteEndpointAwareJdkSSLOptions; +import com.datastax.driver.core.SSLOptions; + +import org.apache.cassandra.config.EncryptionOptions; +import org.apache.cassandra.security.SSLFactory; + +/** + * Holds the SSL and authentication settings used to connect to target hosts during fqltool replay. + * Note that providing any SSL-related configuration option implicitly enables SSL. + */ +public class ConnectionOptions +{ + private final boolean ssl; + private final SSLOptions sslOptions; + private final String authProviderClass; + + private ConnectionOptions(boolean ssl, SSLOptions sslOptions, String authProviderClass) + { + this.ssl = ssl; + this.sslOptions = sslOptions; + this.authProviderClass = authProviderClass; + } + + public boolean ssl() + { + return ssl; + } + + public SSLOptions sslOptions() + { + return sslOptions; + } + + public String authProviderClass() + { + return authProviderClass; + } + + /** + * Builds the configured AuthProvider: a (String,String) constructor when credentials are present, otherwise a no-arg constructor. + */ + @SuppressWarnings("unchecked") + public AuthProvider instantiateAuthProvider(String user, String password) + { + try + { + Class clazz = (Class) Class.forName(authProviderClass); + + if (user != null && password != null) + return clazz.getConstructor(String.class, String.class).newInstance(user, password); + + return clazz.getDeclaredConstructor().newInstance(); + } + catch (NoSuchMethodException e) + { + throw new RuntimeException("Auth provider " + authProviderClass + " does not support plain text credentials", e); + } + catch (Exception e) + { + throw new RuntimeException("Could not instantiate auth provider: " + authProviderClass, e); + } + } + + public static Builder builder() + { + return new Builder(); + } + + public static class Builder + { + private boolean ssl; + private String truststorePath; + private String truststorePassword; + private String keystorePath; + private String keystorePassword; + private String authProviderClass; + + public Builder withSsl(boolean ssl) + { + this.ssl = ssl; + return this; + } + + public Builder withTruststore(String truststorePath) + { + this.truststorePath = truststorePath; + return this; + } + + public Builder withTruststorePassword(String truststorePassword) + { + this.truststorePassword = truststorePassword; + return this; + } + + public Builder withKeystore(String keystorePath) + { + this.keystorePath = keystorePath; + return this; + } + + public Builder withKeystorePassword(String keystorePassword) + { + this.keystorePassword = keystorePassword; + return this; + } + + public Builder withAuthProviderClass(String authProviderClass) + { + this.authProviderClass = authProviderClass; + return this; + } + + public ConnectionOptions build() + { + if (truststorePassword != null && truststorePath == null) + throw new IllegalArgumentException("--ssl-truststore-password requires --ssl-truststore to be set"); + if (keystorePassword != null && keystorePath == null) + throw new IllegalArgumentException("--ssl-keystore-password requires --ssl-keystore to be set"); + + // any SSL-related option implicitly enables SSL + boolean effectiveSsl = ssl || truststorePath != null || keystorePath != null; + + if (authProviderClass != null) + validateAuthProviderClass(); + + SSLOptions sslOptions = effectiveSsl ? buildSSLOptions() : null; + + return new ConnectionOptions(effectiveSsl, sslOptions, authProviderClass); + } + + private void validateAuthProviderClass() + { + try + { + Class clazz = Class.forName(authProviderClass); + if (!AuthProvider.class.isAssignableFrom(clazz)) + throw new IllegalArgumentException(authProviderClass + " does not implement " + AuthProvider.class.getName()); + } + catch (ClassNotFoundException e) + { + throw new RuntimeException("Could not find auth provider class: " + authProviderClass, e); + } + } + + private SSLOptions buildSSLOptions() + { + try + { + EncryptionOptions.ClientEncryptionOptions.Builder encBuilder = new EncryptionOptions.ClientEncryptionOptions.Builder(); + encBuilder.withEnabled(true); + + if (truststorePath != null) + encBuilder.withTrustStore(truststorePath); + if (truststorePassword != null) + encBuilder.withTrustStorePassword(truststorePassword); + + EncryptionOptions.ClientEncryptionOptions.ClientAuth clientAuth = EncryptionOptions.ClientEncryptionOptions.ClientAuth.NOT_REQUIRED; + if (keystorePath != null) + { + encBuilder.withKeyStore(keystorePath); + clientAuth = EncryptionOptions.ClientEncryptionOptions.ClientAuth.REQUIRED; + } + if (keystorePassword != null) + encBuilder.withKeyStorePassword(keystorePassword); + + EncryptionOptions.ClientEncryptionOptions clientEncOptions = encBuilder.build(); + SSLContext sslContext = SSLFactory.createSSLContext(clientEncOptions, clientAuth); + + return RemoteEndpointAwareJdkSSLOptions.builder() + .withSSLContext(sslContext) + .build(); + } + catch (Exception e) + { + throw new RuntimeException("Could not configure SSL for fqltool replay", e); + } + } + } +} diff --git a/tools/fqltool/src/org/apache/cassandra/fqltool/QueryReplayer.java b/tools/fqltool/src/org/apache/cassandra/fqltool/QueryReplayer.java index a0f030322997..dfb14060d152 100644 --- a/tools/fqltool/src/org/apache/cassandra/fqltool/QueryReplayer.java +++ b/tools/fqltool/src/org/apache/cassandra/fqltool/QueryReplayer.java @@ -76,6 +76,20 @@ public QueryReplayer(Iterator> queryIterator, this(queryIterator, targetHosts, resultPaths, filters, queryFilePathString, new DefaultSessionProvider(), null); } + /** + * Constructor that takes SSL and auth provider settings via ConnectionOptions. + */ + public QueryReplayer(Iterator> queryIterator, + List targetHosts, + List resultPaths, + List> filters, + String queryFilePathString, + ConnectionOptions connectionOptions) + { + this(queryIterator, targetHosts, resultPaths, filters, queryFilePathString, + new DefaultSessionProvider(connectionOptions), null); + } + /** * Constructor public to allow external users to build their own session provider * @@ -186,7 +200,7 @@ public void close() throws IOException resultHandler.close(); } - static class ParsedTargetHost + public static class ParsedTargetHost { final int port; final String user; @@ -201,26 +215,42 @@ static class ParsedTargetHost this.password = password; } - static ParsedTargetHost fromString(String s) + /** + * Masks the password in a target host string so it's never logged or written to result paths. + */ + public static String maskPassword(String target) { - String [] userInfoHostPort = s.split("@"); + int at = target.lastIndexOf('@'); + if (at < 0) + return target; + String userInfo = target.substring(0, at); + int colon = userInfo.indexOf(':'); + if (colon < 0) + return target; + return userInfo.substring(0, colon) + ":*****@" + target.substring(at + 1); + } - String hostPort = null; + static ParsedTargetHost fromString(String s) + { + int at = s.lastIndexOf('@'); + String hostPort; String user = null; String password = null; - if (userInfoHostPort.length == 2) + + if (at >= 0) { - String [] userPassword = userInfoHostPort[0].split(":"); - if (userPassword.length != 2) + String userInfo = s.substring(0, at); + hostPort = s.substring(at + 1); + int colon = userInfo.indexOf(':'); + if (colon < 0) throw new RuntimeException("Username provided but no password"); - hostPort = userInfoHostPort[1]; - user = userPassword[0]; - password = userPassword[1]; + user = userInfo.substring(0, colon); + password = userInfo.substring(colon + 1); } - else if (userInfoHostPort.length == 1) - hostPort = userInfoHostPort[0]; else - throw new RuntimeException("Malformed target host: "+s); + { + hostPort = s; + } String[] splitHostPort = hostPort.split(":"); int port = 9042; @@ -241,6 +271,18 @@ private static final class DefaultSessionProvider implements SessionProvider { private final static Map sessionCache = new HashMap<>(); + private final ConnectionOptions connectionOptions; + + DefaultSessionProvider() + { + this(ConnectionOptions.builder().build()); + } + + DefaultSessionProvider(ConnectionOptions connectionOptions) + { + this.connectionOptions = connectionOptions; + } + public synchronized Session connect(String connectionString) { if (sessionCache.containsKey(connectionString)) @@ -249,8 +291,15 @@ public synchronized Session connect(String connectionString) ParsedTargetHost pth = ParsedTargetHost.fromString(connectionString); builder.addContactPoint(pth.host); builder.withPort(pth.port); - if (pth.user != null) + + if (connectionOptions.sslOptions() != null) + builder.withSSL(connectionOptions.sslOptions()); + + if (connectionOptions.authProviderClass() != null) + builder.withAuthProvider(connectionOptions.instantiateAuthProvider(pth.user, pth.password)); + else if (pth.user != null) builder.withCredentials(pth.user, pth.password); + Cluster c = builder.build(); sessionCache.put(connectionString, c.connect()); return sessionCache.get(connectionString); diff --git a/tools/fqltool/src/org/apache/cassandra/fqltool/ResultHandler.java b/tools/fqltool/src/org/apache/cassandra/fqltool/ResultHandler.java index ebc94f874015..29ea364d2301 100644 --- a/tools/fqltool/src/org/apache/cassandra/fqltool/ResultHandler.java +++ b/tools/fqltool/src/org/apache/cassandra/fqltool/ResultHandler.java @@ -47,7 +47,7 @@ public ResultHandler(List targetHosts, List resultPaths, File quer public ResultHandler(List targetHosts, List resultPaths, File queryFilePath, MismatchListener mismatchListener) { - this.targetHosts = targetHosts; + this.targetHosts = targetHosts.stream().map(QueryReplayer.ParsedTargetHost::maskPassword).collect(Collectors.toList()); resultStore = resultPaths != null ? new ResultStore(resultPaths, queryFilePath) : null; resultComparator = new ResultComparator(mismatchListener); } diff --git a/tools/fqltool/src/org/apache/cassandra/fqltool/commands/Replay.java b/tools/fqltool/src/org/apache/cassandra/fqltool/commands/Replay.java index 39b72e3e4b05..3e18eb79d6c1 100644 --- a/tools/fqltool/src/org/apache/cassandra/fqltool/commands/Replay.java +++ b/tools/fqltool/src/org/apache/cassandra/fqltool/commands/Replay.java @@ -34,6 +34,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.fqltool.ConnectionOptions; import org.apache.cassandra.fqltool.FQLQuery; import org.apache.cassandra.fqltool.FQLQueryIterator; import org.apache.cassandra.fqltool.QueryReplayer; @@ -55,7 +56,7 @@ public class Replay implements Runnable @Parameters(paramLabel = "path", description = "Paths containing the full query logs to replay.", arity = "1..*") private List arguments = new ArrayList<>(); - @Option(paramLabel = "target", names = { "--target" }, description = "Hosts to replay the logs to, can be repeated to replay to more hosts.", required = true) + @Option(paramLabel = "target", names = { "--target" }, description = "Hosts to replay the logs to, can be repeated to replay to more hosts. Format: [username:password@]host[:port], for example 'cassandra:cassandra@127.0.0.1:9042'.", required = true) private List targetHosts; @Option(paramLabel = "results", names = { "--results" }, description = "Where to store the results of the queries, this should be a directory. Leave this option out to avoid storing results.") @@ -70,6 +71,24 @@ public class Replay implements Runnable @Option(paramLabel = "replay_ddl_statements", names = { "--replay-ddl-statements" }, description = "If specified, replays DDL statements as well, they are excluded from replaying by default.") private boolean replayDDLStatements; + @Option(paramLabel = "ssl", names = { "--ssl" }, description = "Use SSL for connecting to the target hosts.") + private boolean ssl; + + @Option(paramLabel = "ssl_truststore", names = { "--ssl-truststore" }, description = "Path to the SSL truststore.") + private String truststorePath; + + @Option(paramLabel = "ssl_truststore_password", names = { "--ssl-truststore-password" }, description = "Password for the SSL truststore.") + private String truststorePassword; + + @Option(paramLabel = "ssl_keystore", names = { "--ssl-keystore" }, description = "Path to the SSL keystore, required for two-way SSL.") + private String keystorePath; + + @Option(paramLabel = "ssl_keystore_password", names = { "--ssl-keystore-password" }, description = "Password for the SSL keystore.") + private String keystorePassword; + + @Option(paramLabel = "auth_provider", names = { "--auth-provider" }, description = "Fully qualified class name of a custom com.datastax.driver.core.AuthProvider implementation (e.g. for Kerberos).") + private String authProviderClass; + @Override public void run() { @@ -84,14 +103,24 @@ public void run() System.err.println("The results path (" + basePath + ") should be an existing directory"); throw new IllegalArgumentException("The results path (" + basePath + ") should be an existing directory"); } - resultPaths = targetHosts.stream().map(target -> new File(basePath, target)).collect(Collectors.toList()); + resultPaths = targetHosts.stream().map(target -> new File(basePath, QueryReplayer.ParsedTargetHost.maskPassword(target))).collect(Collectors.toList()); resultPaths.forEach(File::mkdir); } if (targetHosts.size() < 1) { throw new IllegalArgumentException("You need to state at least one --target host to replay the query against"); } - replay(keyspace, arguments, targetHosts, resultPaths, queryStorePath, replayDDLStatements); + + ConnectionOptions connectionOptions = ConnectionOptions.builder() + .withSsl(ssl) + .withTruststore(truststorePath) + .withTruststorePassword(truststorePassword) + .withKeystore(keystorePath) + .withKeystorePassword(keystorePassword) + .withAuthProviderClass(authProviderClass) + .build(); + + replay(keyspace, arguments, targetHosts, resultPaths, queryStorePath, replayDDLStatements, connectionOptions); } catch (Exception e) { @@ -100,6 +129,12 @@ public void run() } public static void replay(String keyspace, List arguments, List targetHosts, List resultPaths, String queryStorePath, boolean replayDDLStatements) + { + replay(keyspace, arguments, targetHosts, resultPaths, queryStorePath, replayDDLStatements, ConnectionOptions.builder().build()); + } + + public static void replay(String keyspace, List arguments, List targetHosts, List resultPaths, String queryStorePath, boolean replayDDLStatements, + ConnectionOptions connectionOptions) { int readAhead = 200; // how many fql queries should we read in to memory to be able to sort them? List readQueues = null; @@ -124,7 +159,7 @@ public static void replay(String keyspace, List arguments, List readQueues = arguments.stream().map(s -> SingleChronicleQueueBuilder.single(s).readOnly(true).build()).collect(Collectors.toList()); iterators = readQueues.stream().map(ChronicleQueue::createTailer).map(tailer -> new FQLQueryIterator(tailer, readAhead)).collect(Collectors.toList()); try (MergeIterator> iter = MergeIterator.get(iterators, FQLQuery::compareTo, new Reducer()); - QueryReplayer replayer = new QueryReplayer(iter, targetHosts, resultPaths, filters, queryStorePath)) + QueryReplayer replayer = new QueryReplayer(iter, targetHosts, resultPaths, filters, queryStorePath, connectionOptions)) { replayer.replay(); } diff --git a/tools/fqltool/test/unit/org/apache/cassandra/fqltool/FQLReplayTest.java b/tools/fqltool/test/unit/org/apache/cassandra/fqltool/FQLReplayTest.java index 734029330add..c0171da15610 100644 --- a/tools/fqltool/test/unit/org/apache/cassandra/fqltool/FQLReplayTest.java +++ b/tools/fqltool/test/unit/org/apache/cassandra/fqltool/FQLReplayTest.java @@ -578,6 +578,125 @@ public void testParser() { assertEquals(9042, pth.port ); assertEquals("aaa", pth.user); assertEquals("bbb", pth.password); + + pth = fromString("user:p@ssword@127.0.0.1:9042"); + assertEquals("127.0.0.1", pth.host); + assertEquals(9042, pth.port); + assertEquals("user", pth.user); + assertEquals("p@ssword", pth.password); + } + + @Test + public void testInvalidAuthProviderClassThrows() + { + try + { + ConnectionOptions.builder().withAuthProviderClass("com.not.a.real.AuthProviderClass").build(); + throw new AssertionError("Expected RuntimeException for invalid auth provider class"); + } + catch (RuntimeException e) + { + assertTrue(e.getMessage().contains("com.not.a.real.AuthProviderClass")); + } + } + + @Test + public void testMaskPassword() + { + assertEquals("aaa:*****@127.0.0.1:9042", QueryReplayer.ParsedTargetHost.maskPassword("aaa:bbb@127.0.0.1:9042")); + assertEquals("127.0.0.1:9042", QueryReplayer.ParsedTargetHost.maskPassword("127.0.0.1:9042")); + assertEquals("127.0.0.1", QueryReplayer.ParsedTargetHost.maskPassword("127.0.0.1")); + assertEquals("user:*****@host:9042", QueryReplayer.ParsedTargetHost.maskPassword("user:p@ssword@host:9042")); + assertEquals("cassandra:*****@127.0.0.1", QueryReplayer.ParsedTargetHost.maskPassword("cassandra:p@ss@w@rd@127.0.0.1")); + } + + @Test + public void testInvalidTruststorePathThrows() + { + try + { + ConnectionOptions.builder().withSsl(true).withTruststore("/path/does/not/exist.jks").withTruststorePassword("password").build(); + throw new AssertionError("Expected RuntimeException for a bad truststore path"); + } + catch (RuntimeException e) + { + assertTrue(e.getMessage().contains("SSL")); + } + } + + @Test + public void testImplicitSslEnableWithTruststoreOnly() + { + // no --ssl flag given, but a truststore path should still turn SSL on rather than being ignored + try + { + ConnectionOptions.builder().withTruststore("/path/does/not/exist.jks").build(); + throw new AssertionError("Expected RuntimeException since SSL should be implicitly enabled"); + } + catch (RuntimeException e) + { + assertTrue(e.getMessage().contains("SSL")); + } + } + + @Test + public void testTruststorePasswordWithoutPathThrows() + { + try + { + ConnectionOptions.builder().withTruststorePassword("somepassword").build(); + throw new AssertionError("Expected IllegalArgumentException when truststore password is given without a path"); + } + catch (IllegalArgumentException e) + { + assertTrue(e.getMessage().contains("--ssl-truststore-password requires --ssl-truststore")); + } + } + + @Test + public void testKeystorePasswordWithoutPathThrows() + { + try + { + ConnectionOptions.builder().withKeystorePassword("somepassword").build(); + throw new AssertionError("Expected IllegalArgumentException when keystore password is given without a path"); + } + catch (IllegalArgumentException e) + { + assertTrue(e.getMessage().contains("--ssl-keystore-password requires --ssl-keystore")); + } + } + + public static class NoArgOnlyAuthProvider implements com.datastax.driver.core.AuthProvider + { + public NoArgOnlyAuthProvider() { } + + public com.datastax.driver.core.Authenticator newAuthenticator(java.net.InetSocketAddress host, String authenticator) + { + throw new UnsupportedOperationException("not needed for this test"); + } + } + + @Test + public void testAuthProviderWithoutCredentialsConstructorThrows() + { + ConnectionOptions connectionOptions = ConnectionOptions.builder() + .withAuthProviderClass(NoArgOnlyAuthProvider.class.getName()) + .build(); + try + { + new QueryReplayer(Collections.emptyIterator(), + Lists.newArrayList("aaa:bbb@127.0.0.1:9999"), + null, + new ArrayList<>(), + null, + connectionOptions); + throw new AssertionError("Expected RuntimeException when auth provider lacks a (String,String) constructor"); + } + catch (RuntimeException e) + { + assertTrue(e.getMessage().contains("does not support plain text credentials")); + } } @Test(expected = RuntimeException.class)