aboutsummaryrefslogtreecommitdiff
path: root/exec/java-exec/src/main/java/org/apache/drill/exec
diff options
context:
space:
mode:
authorParth Chandra <parthc@apache.org>2017-06-09 22:03:59 -0700
committerParth Chandra <parthc@apache.org>2017-10-11 19:26:13 -0700
commit552d7d825e37d42835bd6bfccfc07fc7d3b5fd94 (patch)
tree669a74d1646c97522969f8acf8c4ed2baff8a68f /exec/java-exec/src/main/java/org/apache/drill/exec
parentb803405c0f978beb8beaf77211f99731014ac92f (diff)
DRILL-5431: SSL Support (Java) - Java client server SSL implementation
Also enable OpenSSL support Also fix exclusions and java-exec pom file to eliminate netty-tcnative as a transitive dependency on all projects
Diffstat (limited to 'exec/java-exec/src/main/java/org/apache/drill/exec')
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java28
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/SSLConfig.java69
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java6
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java356
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserConnectionConfig.java13
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java49
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java12
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfig.java265
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigBuilder.java94
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigClient.java273
-rw-r--r--exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigServer.java330
11 files changed, 1259 insertions, 236 deletions
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
index 9b125cb03..5c19e1398 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
@@ -117,6 +117,18 @@ public final class ExecConstants {
public static final String HASHAGG_FALLBACK_ENABLED_KEY = "drill.exec.hashagg.fallback.enabled";
public static final BooleanValidator HASHAGG_FALLBACK_ENABLED_VALIDATOR = new BooleanValidator(HASHAGG_FALLBACK_ENABLED_KEY);
+ public static final String SSL_PROVIDER = "drill.exec.ssl.provider"; // valid values are "JDK", "OPENSSL" // default JDK
+ public static final String SSL_PROTOCOL = "drill.exec.ssl.protocol"; // valid values are SSL, SSLV2, SSLV3, TLS, TLSV1, TLSv1.1, TLSv1.2(default)
+ public static final String SSL_KEYSTORE_TYPE = "drill.exec.ssl.keyStoreType";
+ public static final String SSL_KEYSTORE_PATH = "drill.exec.ssl.keyStorePath"; // path to keystore. default : $JRE_HOME/lib/security/keystore.jks
+ public static final String SSL_KEYSTORE_PASSWORD = "drill.exec.ssl.keyStorePassword"; // default: changeit
+ public static final String SSL_KEY_PASSWORD = "drill.exec.ssl.keyPassword"; //
+ public static final String SSL_TRUSTSTORE_TYPE = "drill.exec.ssl.trustStoreType"; // valid values are jks(default), jceks, pkcs12
+ public static final String SSL_TRUSTSTORE_PATH = "drill.exec.ssl.trustStorePath"; // path to keystore. default : $JRE_HOME/lib/security/cacerts.jks
+ public static final String SSL_TRUSTSTORE_PASSWORD = "drill.exec.ssl.trustStorePassword"; // default: changeit
+ public static final String SSL_USE_HADOOP_CONF = "drill.exec.ssl.useHadoopConfig"; // Initialize ssl params from hadoop if not provided by drill. default: true
+ public static final String SSL_HANDSHAKE_TIMEOUT = "drill.exec.security.user.encryption.ssl.handshakeTimeout"; // Default 10 seconds
+
public static final String TEXT_LINE_READER_BATCH_SIZE = "drill.exec.storage.file.text.batch.size";
public static final String TEXT_LINE_READER_BUFFER_SIZE = "drill.exec.storage.file.text.buffer.size";
public static final String HAZELCAST_SUBNETS = "drill.exec.cache.hazel.subnets";
@@ -133,10 +145,10 @@ public final class ExecConstants {
public static final String HTTP_SESSION_MEMORY_RESERVATION = "drill.exec.http.session.memory.reservation";
public static final String HTTP_SESSION_MEMORY_MAXIMUM = "drill.exec.http.session.memory.maximum";
public static final String HTTP_SESSION_MAX_IDLE_SECS = "drill.exec.http.session_max_idle_secs";
- public static final String HTTP_KEYSTORE_PATH = "drill.exec.ssl.keyStorePath";
- public static final String HTTP_KEYSTORE_PASSWORD = "drill.exec.ssl.keyStorePassword";
- public static final String HTTP_TRUSTSTORE_PATH = "drill.exec.ssl.trustStorePath";
- public static final String HTTP_TRUSTSTORE_PASSWORD = "drill.exec.ssl.trustStorePassword";
+ public static final String HTTP_KEYSTORE_PATH = SSL_KEYSTORE_PATH;
+ public static final String HTTP_KEYSTORE_PASSWORD = SSL_KEYSTORE_PASSWORD;
+ public static final String HTTP_TRUSTSTORE_PATH = SSL_TRUSTSTORE_PATH;
+ public static final String HTTP_TRUSTSTORE_PASSWORD = SSL_TRUSTSTORE_PASSWORD;
public static final String SYS_STORE_PROVIDER_CLASS = "drill.exec.sys.store.provider.class";
public static final String SYS_STORE_PROVIDER_LOCAL_PATH = "drill.exec.sys.store.provider.local.path";
public static final String SYS_STORE_PROVIDER_LOCAL_ENABLE_WRITE = "drill.exec.sys.store.provider.local.write";
@@ -153,6 +165,8 @@ public final class ExecConstants {
public static final String USE_LOGIN_PRINCIPAL = "drill.exec.security.bit.auth.use_login_principal";
public static final String USER_ENCRYPTION_SASL_ENABLED = "drill.exec.security.user.encryption.sasl.enabled";
public static final String USER_ENCRYPTION_SASL_MAX_WRAPPED_SIZE = "drill.exec.security.user.encryption.sasl.max_wrapped_size";
+
+ public static final String USER_SSL_ENABLED = "drill.exec.security.user.encryption.ssl.enabled";
public static final String BIT_ENCRYPTION_SASL_ENABLED = "drill.exec.security.bit.encryption.sasl.enabled";
public static final String BIT_ENCRYPTION_SASL_MAX_WRAPPED_SIZE = "drill.exec.security.bit.encryption.sasl.max_wrapped_size";
@@ -208,7 +222,7 @@ public final class ExecConstants {
public static final OptionValidator PARQUET_DICT_PAGE_SIZE_VALIDATOR = new PositiveLongValidator(PARQUET_DICT_PAGE_SIZE, Integer.MAX_VALUE);
public static final String PARQUET_WRITER_COMPRESSION_TYPE = "store.parquet.compression";
public static final OptionValidator PARQUET_WRITER_COMPRESSION_TYPE_VALIDATOR = new EnumeratedStringValidator(
- PARQUET_WRITER_COMPRESSION_TYPE, "snappy", "gzip", "none");
+ PARQUET_WRITER_COMPRESSION_TYPE, "snappy", "gzip", "none");
public static final String PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING = "store.parquet.enable_dictionary_encoding";
public static final OptionValidator PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING_VALIDATOR = new BooleanValidator(
PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING);
@@ -464,7 +478,7 @@ public final class ExecConstants {
*/
public static final String IMPERSONATION_POLICIES_KEY = "exec.impersonation.inbound_policies";
public static final StringValidator IMPERSONATION_POLICY_VALIDATOR =
- new InboundImpersonationManager.InboundImpersonationPolicyValidator(IMPERSONATION_POLICIES_KEY);
+ new InboundImpersonationManager.InboundImpersonationPolicyValidator(IMPERSONATION_POLICIES_KEY);
/**
@@ -546,5 +560,5 @@ public final class ExecConstants {
public static String bootDefaultFor(String name) {
return OPTION_DEFAULTS_ROOT + name;
- }
+}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/SSLConfig.java b/exec/java-exec/src/main/java/org/apache/drill/exec/SSLConfig.java
deleted file mode 100644
index c6d637480..000000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/SSLConfig.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.drill.exec;
-
-import com.typesafe.config.Config;
-import org.apache.drill.common.exceptions.DrillException;
-
-public class SSLConfig {
-
- private final String keystorePath;
-
- private final String keystorePassword;
-
- private final String truststorePath;
-
- private final String truststorePassword;
-
-
- public SSLConfig(Config config) throws DrillException {
-
- keystorePath = config.getString(ExecConstants.HTTP_KEYSTORE_PATH).trim();
-
- keystorePassword = config.getString(ExecConstants.HTTP_KEYSTORE_PASSWORD).trim();
-
- truststorePath = config.getString(ExecConstants.HTTP_TRUSTSTORE_PATH).trim();
-
- truststorePassword = config.getString(ExecConstants.HTTP_TRUSTSTORE_PASSWORD).trim();
-
- /*If keystorePath or keystorePassword is provided in the configuration file use that*/
- if (!keystorePath.isEmpty() || !keystorePassword.isEmpty()) {
- if (keystorePath.isEmpty()) {
- throw new DrillException(" *.ssl.keyStorePath in the configuration file is empty, but *.ssl.keyStorePassword is set");
- }
- else if (keystorePassword.isEmpty()){
- throw new DrillException(" *.ssl.keyStorePassword in the configuration file is empty, but *.ssl.keyStorePath is set ");
- }
-
- }
- }
-
- public boolean isSslValid() { return !keystorePath.isEmpty() && !keystorePassword.isEmpty(); }
-
- public String getKeyStorePath() { return keystorePath; }
-
- public String getKeyStorePassword() { return keystorePassword; }
-
- public boolean hasTrustStorePath() { return !truststorePath.isEmpty(); }
-
- public boolean hasTrustStorePassword() { return !truststorePassword.isEmpty(); }
-
- public String getTrustStorePath() { return truststorePath; }
-
- public String getTrustStorePassword() { return truststorePassword; }
-} \ No newline at end of file
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
index 9fbbfddaf..84b34a7ca 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
@@ -367,8 +367,12 @@ public class DrillClient implements Closeable, ConnectionThrottle {
DrillbitEndpoint endpoint;
while (triedEndpointIndex < connectTriesVal) {
- client = new UserClient(clientName, config, supportComplexTypes, allocator, eventLoopGroup, executor);
endpoint = endpoints.get(triedEndpointIndex);
+ // Note: the properties member is a DrillProperties instance which lower cases names of
+ // properties. That does not work too well with properties that are mixed case.
+ // For user client severla properties are mixed case so we do not use the properties member
+ // but instead pass the props parameter.
+ client = new UserClient(clientName, config, props, supportComplexTypes, allocator, eventLoopGroup, executor, endpoint);
logger.debug("Connecting to server {}:{}", endpoint.getAddress(), endpoint.getUserPort());
if (!properties.containsKey(DrillProperties.SERVICE_HOST)) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
index 2f4753857..99614bdfa 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
@@ -20,16 +20,25 @@ package org.apache.drill.exec.rpc.user;
import java.io.IOException;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import javax.net.ssl.SSLEngine;
import javax.security.sasl.SaslClient;
import javax.security.sasl.SaslException;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.ssl.SslHandler;
import org.apache.drill.common.KerberosUtil;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.config.DrillProperties;
+import org.apache.drill.common.exceptions.DrillException;
+import org.apache.drill.exec.client.InvalidConnectionInfoException;
+import org.apache.drill.exec.ssl.SSLConfig;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
@@ -55,6 +64,7 @@ import org.apache.drill.exec.proto.UserProtos.UserToBitHandshake;
import org.apache.drill.exec.rpc.AbstractClientConnection;
import org.apache.drill.exec.rpc.Acks;
import org.apache.drill.exec.rpc.BasicClient;
+import org.apache.drill.exec.rpc.ConnectionMultiListener;
import org.apache.drill.exec.rpc.DrillRpcFuture;
import org.apache.drill.exec.rpc.NonTransientRpcException;
import org.apache.drill.exec.rpc.OutOfMemoryHandler;
@@ -62,6 +72,7 @@ import org.apache.drill.exec.rpc.ProtobufLengthDecoder;
import org.apache.drill.exec.rpc.Response;
import org.apache.drill.exec.rpc.ResponseSender;
import org.apache.drill.exec.rpc.RpcConnectionHandler;
+import org.apache.drill.exec.rpc.RpcConstants;
import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.rpc.RpcOutcomeListener;
import org.apache.drill.exec.rpc.security.AuthStringUtil;
@@ -70,7 +81,9 @@ import org.apache.drill.exec.rpc.security.AuthenticatorFactory;
import org.apache.drill.exec.rpc.security.ClientAuthenticatorProvider;
import org.apache.drill.exec.rpc.security.plain.PlainFactory;
import org.apache.drill.exec.rpc.security.SaslProperties;
+import org.apache.drill.exec.ssl.SSLConfigBuilder;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.ssl.SSLFactory;
import org.slf4j.Logger;
import com.google.common.base.Strings;
@@ -87,8 +100,8 @@ import io.netty.buffer.ByteBuf;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
-public class UserClient extends BasicClient<RpcType, UserClient.UserToBitConnection,
- UserToBitHandshake, BitToUserHandshake> {
+public class UserClient
+ extends BasicClient<RpcType, UserClient.UserToBitConnection, UserToBitHandshake, BitToUserHandshake> {
private static final Logger logger = org.slf4j.LoggerFactory.getLogger(UserClient.class);
private final BufferAllocator allocator;
@@ -102,19 +115,47 @@ public class UserClient extends BasicClient<RpcType, UserClient.UserToBitConnect
// these are used for authentication
private volatile List<String> serverAuthMechanisms = null;
private volatile boolean authComplete = true;
-
- public UserClient(String clientName, DrillConfig config, boolean supportComplexTypes,
- BufferAllocator allocator, EventLoopGroup eventLoopGroup, Executor eventExecutor) {
- super(
- UserRpcConfig.getMapping(config, eventExecutor),
- allocator.getAsByteBufAllocator(),
- eventLoopGroup,
- RpcType.HANDSHAKE,
- BitToUserHandshake.class,
- BitToUserHandshake.PARSER);
+ private SSLConfig sslConfig;
+ private DrillbitEndpoint endpoint;
+
+ public UserClient(String clientName, DrillConfig config, Properties properties, boolean supportComplexTypes,
+ BufferAllocator allocator, EventLoopGroup eventLoopGroup, Executor eventExecutor,
+ DrillbitEndpoint endpoint) throws NonTransientRpcException {
+ super(UserRpcConfig.getMapping(config, eventExecutor), allocator.getAsByteBufAllocator(),
+ eventLoopGroup, RpcType.HANDSHAKE, BitToUserHandshake.class, BitToUserHandshake.PARSER);
+ this.endpoint = endpoint; // save the endpoint; it might be needed by SSL init.
this.clientName = clientName;
this.allocator = allocator;
this.supportComplexTypes = supportComplexTypes;
+ try {
+ this.sslConfig = new SSLConfigBuilder().properties(properties).mode(SSLFactory.Mode.CLIENT)
+ .initializeSSLContext(true).validateKeyStore(false).build();
+ } catch (DrillException e) {
+ throw new InvalidConnectionInfoException(e.getMessage());
+ }
+
+ }
+
+ @Override protected void setupSSL(ChannelPipeline pipe,
+ ConnectionMultiListener.SSLHandshakeListener sslHandshakeListener) {
+
+ String peerHost = endpoint.getAddress();
+ int peerPort = endpoint.getUserPort();
+ SSLEngine sslEngine = sslConfig.createSSLEngine(allocator, peerHost, peerPort);
+
+ // Add SSL handler into pipeline
+ SslHandler sslHandler = new SslHandler(sslEngine);
+ sslHandler.setHandshakeTimeoutMillis(sslConfig.getHandshakeTimeout());
+
+ // Add a listener for SSL Handshake complete. The Drill client handshake will be enabled only
+ // after this is done.
+ sslHandler.handshakeFuture().addListener(sslHandshakeListener);
+ pipe.addFirst(RpcConstants.SSL_HANDLER, sslHandler);
+ logger.debug(sslConfig.toString());
+ }
+
+ @Override protected boolean isSslEnabled() {
+ return sslConfig.isUserSslEnabled();
}
public RpcEndpointInfos getServerInfos() {
@@ -132,38 +173,51 @@ public class UserClient extends BasicClient<RpcType, UserClient.UserToBitConnect
/**
* Connects, and if required, authenticates. This method blocks until both operations are complete.
*
- * @param endpoint endpoint to connect to
- * @param properties properties
+ * @param endpoint endpoint to connect to
+ * @param properties properties
* @param credentials credentials
* @throws RpcException if either connection or authentication fails
*/
public void connect(final DrillbitEndpoint endpoint, final DrillProperties properties,
- final UserCredentials credentials) throws RpcException {
- final UserToBitHandshake.Builder hsBuilder = UserToBitHandshake.newBuilder()
- .setRpcVersion(UserRpcConfig.RPC_VERSION)
- .setSupportListening(true)
- .setSupportComplexTypes(supportComplexTypes)
- .setSupportTimeout(true)
- .setCredentials(credentials)
- .setClientInfos(UserRpcUtils.getRpcEndpointInfos(clientName))
- .setSaslSupport(SaslSupport.SASL_PRIVACY)
- .setProperties(properties.serializeForServer());
+ final UserCredentials credentials) throws RpcException {
+ final UserToBitHandshake.Builder hsBuilder =
+ UserToBitHandshake.newBuilder()
+ .setRpcVersion(UserRpcConfig.RPC_VERSION)
+ .setSupportListening(true)
+ .setSupportComplexTypes(supportComplexTypes)
+ .setSupportTimeout(true).setCredentials(credentials)
+ .setClientInfos(UserRpcUtils.getRpcEndpointInfos(clientName))
+ .setSaslSupport(SaslSupport.SASL_PRIVACY)
+ .setProperties(properties.serializeForServer());
// Only used for testing purpose
if (properties.containsKey(DrillProperties.TEST_SASL_LEVEL)) {
- hsBuilder.setSaslSupport(SaslSupport.valueOf(
- Integer.parseInt(properties.getProperty(DrillProperties.TEST_SASL_LEVEL))));
+ hsBuilder.setSaslSupport(
+ SaslSupport.valueOf(Integer.parseInt(properties.getProperty(DrillProperties.TEST_SASL_LEVEL))));
}
- connect(hsBuilder.build(), endpoint).checkedGet();
+ if (sslConfig.isUserSslEnabled()) {
+ try {
+ connect(hsBuilder.build(), endpoint)
+ .checkedGet(sslConfig.getHandshakeTimeout(), TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ String msg = new StringBuilder().append(
+ "Connecting to the server timed out. This is sometimes due to a mismatch in the SSL configuration between client and server. [ Exception: ")
+ .append(e.getMessage()).append("]").toString();
+ throw new NonTransientRpcException(msg);
+ }
+ } else {
+ connect(hsBuilder.build(), endpoint).checkedGet();
+ }
// Check if client needs encryption and server is not configured for encryption.
- final boolean clientNeedsEncryption = properties.containsKey(DrillProperties.SASL_ENCRYPT)
- && Boolean.parseBoolean(properties.getProperty(DrillProperties.SASL_ENCRYPT));
+ final boolean clientNeedsEncryption = properties.containsKey(DrillProperties.SASL_ENCRYPT) && Boolean
+ .parseBoolean(properties.getProperty(DrillProperties.SASL_ENCRYPT));
- if(clientNeedsEncryption && !connection.isEncryptionEnabled()) {
- throw new NonTransientRpcException("Client needs encrypted connection but server is not configured for " +
- "encryption. Please check connection parameter or contact your administrator");
+ if (clientNeedsEncryption && !connection.isEncryptionEnabled()) {
+ throw new NonTransientRpcException(
+ "Client needs encrypted connection but server is not configured for "
+ + "encryption. Please check connection parameter or contact your administrator");
}
if (serverAuthMechanisms != null) {
@@ -176,31 +230,28 @@ public class UserClient extends BasicClient<RpcType, UserClient.UserToBitConnect
}
private CheckedFuture<Void, RpcException> connect(final UserToBitHandshake handshake,
- final DrillbitEndpoint endpoint) {
+ final DrillbitEndpoint endpoint) {
final SettableFuture<Void> connectionSettable = SettableFuture.create();
final CheckedFuture<Void, RpcException> connectionFuture =
new AbstractCheckedFuture<Void, RpcException>(connectionSettable) {
- @Override
- protected RpcException mapException(Exception e) {
+ @Override protected RpcException mapException(Exception e) {
return RpcException.mapException(e);
}
};
final RpcConnectionHandler<UserToBitConnection> connectionHandler =
new RpcConnectionHandler<UserToBitConnection>() {
- @Override
- public void connectionSucceeded(UserToBitConnection connection) {
+ @Override public void connectionSucceeded(UserToBitConnection connection) {
connectionSettable.set(null);
}
- @Override
- public void connectionFailed(FailureType type, Throwable t) {
- connectionSettable.setException(new RpcException(String.format("%s : %s",
- type.name(), t.getMessage()), t));
+ @Override public void connectionFailed(FailureType type, Throwable t) {
+ connectionSettable
+ .setException(new RpcException(String.format("%s : %s", type.name(), t.getMessage()), t));
}
};
- connectAsClient(queryResultHandler.getWrappedConnectionHandler(connectionHandler),
- handshake, endpoint.getAddress(), endpoint.getUserPort());
+ connectAsClient(queryResultHandler.getWrappedConnectionHandler(connectionHandler), handshake,
+ endpoint.getAddress(), endpoint.getUserPort());
return connectionFuture;
}
@@ -211,15 +262,15 @@ public class UserClient extends BasicClient<RpcType, UserClient.UserToBitConnect
// Set correct QOP property and Strength based on server needs encryption or not.
// If ChunkMode is enabled then negotiate for buffer size equal to wrapChunkSize,
// If ChunkMode is disabled then negotiate for MAX_WRAPPED_SIZE buffer size.
- propertiesMap.putAll(SaslProperties.getSaslProperties(connection.isEncryptionEnabled(),
- connection.getMaxWrappedSize()));
+ propertiesMap.putAll(
+ SaslProperties.getSaslProperties(connection.isEncryptionEnabled(), connection.getMaxWrappedSize()));
- final SettableFuture<Void> authSettable = SettableFuture.create(); // use handleAuthFailure to setException
+ final SettableFuture<Void> authSettable =
+ SettableFuture.create(); // use handleAuthFailure to setException
final CheckedFuture<Void, SaslException> authFuture =
new AbstractCheckedFuture<Void, SaslException>(authSettable) {
- @Override
- protected SaslException mapException(Exception e) {
+ @Override protected SaslException mapException(Exception e) {
if (e instanceof ExecutionException) {
final Throwable cause = Throwables.getRootCause(e);
if (cause instanceof SaslException) {
@@ -227,8 +278,9 @@ public class UserClient extends BasicClient<RpcType, UserClient.UserToBitConnect
connection.getEncryptionCtxtString(), cause.getMessage()), cause);
}
}
- return new SaslException(String.format("Authentication failed unexpectedly. [Details: %s, Error %s]",
- connection.getEncryptionCtxtString(), e.getMessage()), e);
+ return new SaslException(String
+ .format("Authentication failed unexpectedly. [Details: %s, Error %s]",
+ connection.getEncryptionCtxtString(), e.getMessage()), e);
}
};
@@ -244,8 +296,10 @@ public class UserClient extends BasicClient<RpcType, UserClient.UserToBitConnect
ugi = factory.createAndLoginUser(propertiesMap);
saslClient = factory.createSaslClient(ugi, propertiesMap);
if (saslClient == null) {
- throw new SaslException(String.format("Cannot initiate authentication using %s mechanism. Insufficient " +
- "credentials or selected mechanism doesn't support configured security layers?", factory.getSimpleName()));
+ throw new SaslException(String.format(
+ "Cannot initiate authentication using %s mechanism. Insufficient "
+ + "credentials or selected mechanism doesn't support configured security layers?",
+ factory.getSimpleName()));
}
connection.setSaslClient(saslClient);
} catch (final IOException e) {
@@ -256,26 +310,24 @@ public class UserClient extends BasicClient<RpcType, UserClient.UserToBitConnect
logger.trace("Initiating SASL exchange.");
new AuthenticationOutcomeListener<>(this, connection, RpcType.SASL_MESSAGE, ugi,
new RpcOutcomeListener<Void>() {
- @Override
- public void failed(RpcException ex) {
+ @Override public void failed(RpcException ex) {
authSettable.setException(ex);
}
- @Override
- public void success(Void value, ByteBuf buffer) {
+ @Override public void success(Void value, ByteBuf buffer) {
authComplete = true;
authSettable.set(null);
}
- @Override
- public void interrupted(InterruptedException e) {
+ @Override public void interrupted(InterruptedException e) {
authSettable.setException(e);
}
}).initiate(mechanismName);
return authFuture;
}
- private AuthenticatorFactory getAuthenticatorFactory(final DrillProperties properties) throws SaslException {
+ private AuthenticatorFactory getAuthenticatorFactory(final DrillProperties properties)
+ throws SaslException {
final Set<String> mechanismSet = AuthStringUtil.asSet(serverAuthMechanisms);
// first, check if a certain mechanism must be used
@@ -285,129 +337,126 @@ public class UserClient extends BasicClient<RpcType, UserClient.UserToBitConnect
throw new SaslException(String.format("Unknown mechanism: %s", authMechanism));
}
if (!mechanismSet.contains(authMechanism.toUpperCase())) {
- throw new SaslException(String.format("Server does not support authentication using: %s. [Details: %s]",
- authMechanism, connection.getEncryptionCtxtString()));
+ throw new SaslException(String
+ .format("Server does not support authentication using: %s. [Details: %s]", authMechanism,
+ connection.getEncryptionCtxtString()));
}
- return ClientAuthenticatorProvider.getInstance()
- .getAuthenticatorFactory(authMechanism);
+ return ClientAuthenticatorProvider.getInstance().getAuthenticatorFactory(authMechanism);
}
// check if Kerberos is supported, and the service principal is provided
- if (mechanismSet.contains(KerberosUtil.KERBEROS_SIMPLE_NAME) &&
- properties.containsKey(DrillProperties.SERVICE_PRINCIPAL)) {
+ if (mechanismSet.contains(KerberosUtil.KERBEROS_SIMPLE_NAME) && properties
+ .containsKey(DrillProperties.SERVICE_PRINCIPAL)) {
return ClientAuthenticatorProvider.getInstance()
.getAuthenticatorFactory(KerberosUtil.KERBEROS_SIMPLE_NAME);
}
// check if username/password is supported, and username/password are provided
- if (mechanismSet.contains(PlainFactory.SIMPLE_NAME) &&
- properties.containsKey(DrillProperties.USER) &&
- !Strings.isNullOrEmpty(properties.getProperty(DrillProperties.PASSWORD))) {
- return ClientAuthenticatorProvider.getInstance()
- .getAuthenticatorFactory(PlainFactory.SIMPLE_NAME);
+ if (mechanismSet.contains(PlainFactory.SIMPLE_NAME) && properties.containsKey(DrillProperties.USER)
+ && !Strings.isNullOrEmpty(properties.getProperty(DrillProperties.PASSWORD))) {
+ return ClientAuthenticatorProvider.getInstance().getAuthenticatorFactory(PlainFactory.SIMPLE_NAME);
}
- throw new SaslException(String.format("Server requires authentication using %s. Insufficient credentials?. " +
- "[Details: %s]. ", serverAuthMechanisms, connection.getEncryptionCtxtString()));
+ throw new SaslException(String
+ .format("Server requires authentication using %s. Insufficient credentials?. " + "[Details: %s]. ",
+ serverAuthMechanisms, connection.getEncryptionCtxtString()));
}
- protected <SEND extends MessageLite, RECEIVE extends MessageLite>
- void send(RpcOutcomeListener<RECEIVE> listener, RpcType rpcType, SEND protobufBody, Class<RECEIVE> clazz,
- boolean allowInEventLoop, ByteBuf... dataBodies) {
+ protected <SEND extends MessageLite, RECEIVE extends MessageLite> void send(
+ RpcOutcomeListener<RECEIVE> listener, RpcType rpcType, SEND protobufBody, Class<RECEIVE> clazz,
+ boolean allowInEventLoop, ByteBuf... dataBodies) {
super.send(listener, connection, rpcType, protobufBody, clazz, allowInEventLoop, dataBodies);
}
- @Override
- protected MessageLite getResponseDefaultInstance(int rpcType) throws RpcException {
+ @Override protected MessageLite getResponseDefaultInstance(int rpcType) throws RpcException {
switch (rpcType) {
- case RpcType.ACK_VALUE:
- return Ack.getDefaultInstance();
- case RpcType.HANDSHAKE_VALUE:
- return BitToUserHandshake.getDefaultInstance();
- case RpcType.QUERY_HANDLE_VALUE:
- return QueryId.getDefaultInstance();
- case RpcType.QUERY_RESULT_VALUE:
- return QueryResult.getDefaultInstance();
- case RpcType.QUERY_DATA_VALUE:
- return QueryData.getDefaultInstance();
- case RpcType.QUERY_PLAN_FRAGMENTS_VALUE:
- return QueryPlanFragments.getDefaultInstance();
- case RpcType.CATALOGS_VALUE:
- return GetCatalogsResp.getDefaultInstance();
- case RpcType.SCHEMAS_VALUE:
- return GetSchemasResp.getDefaultInstance();
- case RpcType.TABLES_VALUE:
- return GetTablesResp.getDefaultInstance();
- case RpcType.COLUMNS_VALUE:
- return GetColumnsResp.getDefaultInstance();
- case RpcType.PREPARED_STATEMENT_VALUE:
- return CreatePreparedStatementResp.getDefaultInstance();
- case RpcType.SASL_MESSAGE_VALUE:
- return SaslMessage.getDefaultInstance();
- case RpcType.SERVER_META_VALUE:
- return GetServerMetaResp.getDefaultInstance();
+ case RpcType.ACK_VALUE:
+ return Ack.getDefaultInstance();
+ case RpcType.HANDSHAKE_VALUE:
+ return BitToUserHandshake.getDefaultInstance();
+ case RpcType.QUERY_HANDLE_VALUE:
+ return QueryId.getDefaultInstance();
+ case RpcType.QUERY_RESULT_VALUE:
+ return QueryResult.getDefaultInstance();
+ case RpcType.QUERY_DATA_VALUE:
+ return QueryData.getDefaultInstance();
+ case RpcType.QUERY_PLAN_FRAGMENTS_VALUE:
+ return QueryPlanFragments.getDefaultInstance();
+ case RpcType.CATALOGS_VALUE:
+ return GetCatalogsResp.getDefaultInstance();
+ case RpcType.SCHEMAS_VALUE:
+ return GetSchemasResp.getDefaultInstance();
+ case RpcType.TABLES_VALUE:
+ return GetTablesResp.getDefaultInstance();
+ case RpcType.COLUMNS_VALUE:
+ return GetColumnsResp.getDefaultInstance();
+ case RpcType.PREPARED_STATEMENT_VALUE:
+ return CreatePreparedStatementResp.getDefaultInstance();
+ case RpcType.SASL_MESSAGE_VALUE:
+ return SaslMessage.getDefaultInstance();
+ case RpcType.SERVER_META_VALUE:
+ return GetServerMetaResp.getDefaultInstance();
}
throw new RpcException(String.format("Unable to deal with RpcType of %d", rpcType));
}
- @Override
- protected void handle(UserToBitConnection connection, int rpcType, ByteBuf pBody, ByteBuf dBody,
- ResponseSender sender) throws RpcException {
+ @Override protected void handle(UserToBitConnection connection, int rpcType, ByteBuf pBody, ByteBuf dBody,
+ ResponseSender sender) throws RpcException {
if (!authComplete) {
// Remote should not be making any requests before authenticating, drop connection
- throw new RpcException(String.format("Request of type %d is not allowed without authentication. " +
- "Remote on %s must authenticate before making requests. Connection dropped.",
- rpcType, connection.getRemoteAddress()));
+ throw new RpcException(String.format("Request of type %d is not allowed without authentication. "
+ + "Remote on %s must authenticate before making requests. Connection dropped.", rpcType,
+ connection.getRemoteAddress()));
}
switch (rpcType) {
- case RpcType.QUERY_DATA_VALUE:
- queryResultHandler.batchArrived(connection, pBody, dBody);
- sender.send(new Response(RpcType.ACK, Acks.OK));
- break;
- case RpcType.QUERY_RESULT_VALUE:
- queryResultHandler.resultArrived(pBody);
- sender.send(new Response(RpcType.ACK, Acks.OK));
- break;
- default:
- throw new RpcException(String.format("Unknown Rpc Type %d. ", rpcType));
+ case RpcType.QUERY_DATA_VALUE:
+ queryResultHandler.batchArrived(connection, pBody, dBody);
+ sender.send(new Response(RpcType.ACK, Acks.OK));
+ break;
+ case RpcType.QUERY_RESULT_VALUE:
+ queryResultHandler.resultArrived(pBody);
+ sender.send(new Response(RpcType.ACK, Acks.OK));
+ break;
+ default:
+ throw new RpcException(String.format("Unknown Rpc Type %d. ", rpcType));
}
}
- @Override
- protected void validateHandshake(BitToUserHandshake inbound) throws RpcException {
-// logger.debug("Handling handshake from bit to user. {}", inbound);
+ @Override protected void validateHandshake(BitToUserHandshake inbound) throws RpcException {
+ // logger.debug("Handling handshake from bit to user. {}", inbound);
if (inbound.hasServerInfos()) {
serverInfos = inbound.getServerInfos();
}
supportedMethods = Sets.immutableEnumSet(inbound.getSupportedMethodsList());
switch (inbound.getStatus()) {
- case SUCCESS:
- break;
- case AUTH_REQUIRED: {
- authComplete = false;
- serverAuthMechanisms = ImmutableList.copyOf(inbound.getAuthenticationMechanismsList());
- connection.setEncryption(inbound.hasEncrypted() && inbound.getEncrypted());
-
- if (inbound.hasMaxWrappedSize()) {
- connection.setMaxWrappedSize(inbound.getMaxWrappedSize());
+ case SUCCESS:
+ break;
+ case AUTH_REQUIRED: {
+ authComplete = false;
+ serverAuthMechanisms = ImmutableList.copyOf(inbound.getAuthenticationMechanismsList());
+ connection.setEncryption(inbound.hasEncrypted() && inbound.getEncrypted());
+
+ if (inbound.hasMaxWrappedSize()) {
+ connection.setMaxWrappedSize(inbound.getMaxWrappedSize());
+ }
+ logger.trace(String
+ .format("Server requires authentication with encryption context %s before proceeding.",
+ connection.getEncryptionCtxtString()));
+ break;
}
- logger.trace(String.format("Server requires authentication with encryption context %s before proceeding.",
- connection.getEncryptionCtxtString()));
- break;
- }
- case AUTH_FAILED:
- case RPC_VERSION_MISMATCH:
- case UNKNOWN_FAILURE:
- final String errMsg = String.format("Status: %s, Error Id: %s, Error message: %s",
- inbound.getStatus(), inbound.getErrorId(), inbound.getErrorMessage());
- logger.error(errMsg);
- throw new NonTransientRpcException(errMsg);
+ case AUTH_FAILED:
+ case RPC_VERSION_MISMATCH:
+ case UNKNOWN_FAILURE:
+ final String errMsg = String
+ .format("Status: %s, Error Id: %s, Error message: %s", inbound.getStatus(),
+ inbound.getErrorId(), inbound.getErrorMessage());
+ logger.error(errMsg);
+ throw new NonTransientRpcException(errMsg);
}
}
- @Override
- protected UserToBitConnection initRemoteConnection(SocketChannel channel) {
+ @Override protected UserToBitConnection initRemoteConnection(SocketChannel channel) {
super.initRemoteConnection(channel);
return new UserToBitConnection(channel);
}
@@ -421,39 +470,34 @@ public class UserClient extends BasicClient<RpcType, UserClient.UserToBitConnect
super(channel, "user client");
}
- @Override
- public BufferAllocator getAllocator() {
+ @Override public BufferAllocator getAllocator() {
return allocator;
}
- @Override
- protected Logger getLogger() {
+ @Override protected Logger getLogger() {
return logger;
}
- @Override
- public void incConnectionCounter() {
+ @Override public void incConnectionCounter() {
// no-op
}
- @Override
- public void decConnectionCounter() {
+ @Override public void decConnectionCounter() {
// no-op
}
}
- @Override
- public ProtobufLengthDecoder getDecoder(BufferAllocator allocator) {
+ @Override public ProtobufLengthDecoder getDecoder(BufferAllocator allocator) {
return new UserProtobufLengthDecoder(allocator, OutOfMemoryHandler.DEFAULT_INSTANCE);
}
/**
* planQuery is an API to plan a query without query execution
+ *
* @param req - data necessary to plan query
* @return list of PlanFragments that can later on be submitted for execution
*/
- public DrillRpcFuture<QueryPlanFragments> planQuery(
- GetQueryPlanFragments req) {
+ public DrillRpcFuture<QueryPlanFragments> planQuery(GetQueryPlanFragments req) {
return send(RpcType.GET_QUERY_PLAN_FRAGMENTS, req, QueryPlanFragments.class);
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserConnectionConfig.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserConnectionConfig.java
index 57ad4d51b..4e4c80e97 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserConnectionConfig.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserConnectionConfig.java
@@ -33,6 +33,7 @@ class UserConnectionConfig extends AbstractConnectionConfig {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserConnectionConfig.class);
private final boolean authEnabled;
+ private final boolean sslEnabled;
private final InboundImpersonationManager impersonationManager;
private final UserServerRequestHandler handler;
@@ -75,10 +76,16 @@ class UserConnectionConfig extends AbstractConnectionConfig {
} else {
authEnabled = false;
}
-
impersonationManager = !config.getBoolean(ExecConstants.IMPERSONATION_ENABLED)
? null
: new InboundImpersonationManager();
+
+ sslEnabled = config.getBoolean(ExecConstants.USER_SSL_ENABLED);
+
+ if(isSSLEnabled() && isAuthEnabled() && isEncryptionEnabled()){
+ logger.warn("The server is configured to use both SSL and SASL encryption (only one should be configured).");
+ }
+
}
@Override
@@ -90,6 +97,10 @@ class UserConnectionConfig extends AbstractConnectionConfig {
return authEnabled;
}
+ boolean isSSLEnabled() {
+ return sslEnabled;
+ }
+
InboundImpersonationManager getImpersonationManager() {
return impersonationManager;
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
index 254fdca29..c0573db81 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
@@ -21,9 +21,15 @@ import java.io.IOException;
import java.net.SocketAddress;
import java.util.UUID;
+import javax.net.ssl.SSLEngine;
import javax.security.sasl.SaslException;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.ssl.SslHandler;
import org.apache.drill.common.config.DrillProperties;
+import org.apache.drill.common.exceptions.DrillException;
+import org.apache.drill.exec.ssl.SSLConfig;
import org.apache.drill.exec.exception.DrillbitStartupException;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.physical.impl.materialize.QueryWritableBatch;
@@ -53,8 +59,10 @@ import org.apache.drill.exec.rpc.security.plain.PlainFactory;
import org.apache.drill.exec.rpc.user.UserServer.BitToUserConnection;
import org.apache.drill.exec.rpc.user.security.UserAuthenticationException;
import org.apache.drill.exec.server.BootStrapContext;
+import org.apache.drill.exec.ssl.SSLConfigBuilder;
import org.apache.drill.exec.work.user.UserWorker;
import org.apache.hadoop.security.HadoopKerberosName;
+import org.apache.hadoop.security.ssl.SSLFactory;
import org.slf4j.Logger;
import com.google.protobuf.MessageLite;
@@ -71,6 +79,8 @@ public class UserServer extends BasicServer<RpcType, BitToUserConnection> {
private static final String SERVER_NAME = "Apache Drill Server";
private final UserConnectionConfig config;
+ private final SSLConfig sslConfig;
+ private Channel sslChannel;
private final UserWorker userWorker;
public UserServer(BootStrapContext context, BufferAllocator allocator, EventLoopGroup eventLoopGroup,
@@ -79,6 +89,17 @@ public class UserServer extends BasicServer<RpcType, BitToUserConnection> {
allocator.getAsByteBufAllocator(),
eventLoopGroup);
this.config = new UserConnectionConfig(allocator, context, new UserServerRequestHandler(worker));
+ this.sslChannel = null;
+ try {
+ this.sslConfig = new SSLConfigBuilder()
+ .config(context.getConfig())
+ .mode(SSLFactory.Mode.SERVER)
+ .initializeSSLContext(true)
+ .validateKeyStore(true)
+ .build();
+ } catch (DrillException e) {
+ throw new DrillbitStartupException(e.getMessage(), e.getCause());
+ }
this.userWorker = worker;
// Initialize Singleton instance of UserRpcMetrics.
@@ -86,6 +107,34 @@ public class UserServer extends BasicServer<RpcType, BitToUserConnection> {
}
@Override
+ protected void setupSSL(ChannelPipeline pipe) {
+
+ SSLEngine sslEngine = sslConfig.createSSLEngine(config.getAllocator(), null, 0);
+ // Add SSL handler into pipeline
+ pipe.addFirst(RpcConstants.SSL_HANDLER, new SslHandler(sslEngine));
+ logger.debug("SSL communication between client and server is enabled.");
+ logger.debug(sslConfig.toString());
+
+ }
+
+ @Override
+ protected boolean isSslEnabled() {
+ return sslConfig.isUserSslEnabled();
+ }
+
+ @Override
+ public void setSslChannel(Channel c) {
+ sslChannel = c;
+ }
+
+ @Override
+ protected void closeSSL(){
+ if(isSslEnabled() && sslChannel != null){
+ sslChannel.close();
+ }
+ }
+
+ @Override
protected MessageLite getResponseDefaultInstance(int rpcType) throws RpcException {
// a user server only expects acknowledgments on messages it creates.
switch (rpcType) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java
index 69f2cab76..e35f5421e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java
@@ -26,12 +26,14 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.drill.common.exceptions.DrillException;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecConstants;
-import org.apache.drill.exec.SSLConfig;
+import org.apache.drill.exec.ssl.SSLConfig;
import org.apache.drill.exec.exception.DrillbitStartupException;
import org.apache.drill.exec.rpc.security.plain.PlainFactory;
import org.apache.drill.exec.server.BootStrapContext;
import org.apache.drill.exec.server.rest.auth.DrillRestLoginService;
+import org.apache.drill.exec.ssl.SSLConfigBuilder;
import org.apache.drill.exec.work.WorkManager;
+import org.apache.hadoop.security.ssl.SSLFactory;
import org.bouncycastle.asn1.x500.X500NameBuilder;
import org.bouncycastle.asn1.x500.style.BCStyle;
import org.bouncycastle.cert.X509v3CertificateBuilder;
@@ -331,12 +333,18 @@ public class WebServer implements AutoCloseable {
logger.info("Setting up HTTPS connector for web server");
final SslContextFactory sslContextFactory = new SslContextFactory();
- SSLConfig ssl = new SSLConfig(config);
+ SSLConfig ssl = new SSLConfigBuilder()
+ .config(config)
+ .mode(SSLFactory.Mode.SERVER)
+ .initializeSSLContext(false)
+ .validateKeyStore(true)
+ .build();
if(ssl.isSslValid()){
logger.info("Using configured SSL settings for web server");
sslContextFactory.setKeyStorePath(ssl.getKeyStorePath());
sslContextFactory.setKeyStorePassword(ssl.getKeyStorePassword());
+ sslContextFactory.setKeyManagerPassword(ssl.getKeyPassword());
if(ssl.hasTrustStorePath()){
sslContextFactory.setTrustStorePath(ssl.getTrustStorePath());
if(ssl.hasTrustStorePassword()){
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfig.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfig.java
new file mode 100644
index 000000000..98a8c8d4f
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfig.java
@@ -0,0 +1,265 @@
+/*
+ * 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.drill.exec.ssl;
+
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslProvider;
+import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import org.apache.drill.common.exceptions.DrillException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.hadoop.security.ssl.SSLFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.TrustManagerFactory;
+import java.io.FileInputStream;
+import java.io.InputStream;
+import java.security.KeyStore;
+
+public abstract class SSLConfig {
+
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SSLConfig.class);
+
+ public static final String DEFAULT_SSL_PROVIDER = "JDK"; // JDK or OPENSSL
+ public static final String DEFAULT_SSL_PROTOCOL = "TLSv1.2";
+ public static final int DEFAULT_SSL_HANDSHAKE_TIMEOUT_MS = 10 * 1000; // 10 seconds
+
+ // Either the Netty SSL context or the JDK SSL context will be initialized
+ // The JDK SSL context is use iff the useSystemTrustStore setting is enabled.
+ protected SslContext nettySslContext;
+ protected SSLContext jdkSSlContext;
+
+ private static final boolean isWindows = System.getProperty("os.name").toLowerCase().indexOf("win") >= 0;
+ private static final boolean isMacOs = System.getProperty("os.name").toLowerCase().indexOf("mac") >= 0;
+
+ public static final String HADOOP_SSL_CONF_TPL_KEY = "hadoop.ssl.{0}.conf";
+ public static final String HADOOP_SSL_KEYSTORE_LOCATION_TPL_KEY = "ssl.{0}.keystore.location";
+ public static final String HADOOP_SSL_KEYSTORE_PASSWORD_TPL_KEY = "ssl.{0}.keystore.password";
+ public static final String HADOOP_SSL_KEYSTORE_TYPE_TPL_KEY = "ssl.{0}.keystore.type";
+ public static final String HADOOP_SSL_KEYSTORE_KEYPASSWORD_TPL_KEY =
+ "ssl.{0}.keystore.keypassword";
+ public static final String HADOOP_SSL_TRUSTSTORE_LOCATION_TPL_KEY = "ssl.{0}.truststore.location";
+ public static final String HADOOP_SSL_TRUSTSTORE_PASSWORD_TPL_KEY = "ssl.{0}.truststore.password";
+ public static final String HADOOP_SSL_TRUSTSTORE_TYPE_TPL_KEY = "ssl.{0}.truststore.type";
+
+ public SSLConfig() {
+ }
+
+ public abstract void validateKeyStore() throws DrillException;
+
+ // We need to use different SSLContext objects depending on what the user has chosen
+ // For most uses we will use the Netty SslContext class. This allows us to use either
+ // the JDK implementation or the OpenSSL implementation. However if the user wants to
+ // use the system trust store, then the only way to access it is via the JDK's
+ // SSLContext class. (See the createSSLEngine method below).
+
+ public abstract SslContext initNettySslContext() throws DrillException;
+
+ public abstract SSLContext initJDKSSLContext() throws DrillException;
+
+ public abstract boolean isUserSslEnabled();
+
+ public abstract boolean isHttpsEnabled();
+
+ public abstract String getKeyStoreType();
+
+ public abstract String getKeyStorePath();
+
+ public abstract String getKeyStorePassword();
+
+ public abstract String getKeyPassword();
+
+ public abstract String getTrustStoreType();
+
+ public abstract boolean hasTrustStorePath();
+
+ public abstract String getTrustStorePath();
+
+ public abstract boolean hasTrustStorePassword();
+
+ public abstract String getTrustStorePassword();
+
+ public abstract String getProtocol();
+
+ public abstract SslProvider getProvider();
+
+ public abstract int getHandshakeTimeout();
+
+ public abstract SSLFactory.Mode getMode();
+
+ public abstract boolean disableHostVerification();
+
+ public abstract boolean disableCertificateVerification();
+
+ public abstract boolean useSystemTrustStore();
+
+ public abstract boolean isSslValid();
+
+ public SslContext getNettySslContext() {
+ return nettySslContext;
+ }
+
+ public TrustManagerFactory initializeTrustManagerFactory() throws DrillException {
+ TrustManagerFactory tmf;
+ KeyStore ts = null;
+ //Support Windows/MacOs system trust store
+ try {
+ String trustStoreType = getTrustStoreType();
+ if ((isWindows || isMacOs) && useSystemTrustStore()) {
+ // This is valid for MS-Windows and MacOs
+ logger.debug("Initializing System truststore.");
+ ts = KeyStore.getInstance(!trustStoreType.isEmpty() ? trustStoreType : KeyStore.getDefaultType());
+ ts.load(null, null);
+ } else if (!getTrustStorePath().isEmpty()) {
+ // if truststore is not provided then we will use the default. Note that the default depends on
+ // the TrustManagerFactory that in turn depends on the Security Provider.
+ // Use null as the truststore which will result in the default truststore being picked up
+ logger.debug("Initializing truststore {}.", getTrustStorePath());
+ ts = KeyStore.getInstance(!trustStoreType.isEmpty() ? trustStoreType : KeyStore.getDefaultType());
+ InputStream tsStream = new FileInputStream(getTrustStorePath());
+ ts.load(tsStream, getTrustStorePassword().toCharArray());
+ } else {
+ logger.debug("Initializing default truststore.");
+ }
+ if (disableCertificateVerification()) {
+ tmf = InsecureTrustManagerFactory.INSTANCE;
+ } else {
+ tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ }
+ tmf.init(ts);
+ } catch (Exception e) {
+ // Catch any SSL initialization Exceptions here and abort.
+ throw new DrillException(
+ new StringBuilder()
+ .append("Exception while initializing the truststore: [")
+ .append(e.getMessage())
+ .append("]. ")
+ .toString(), e);
+ }
+ return tmf;
+ }
+
+ public KeyManagerFactory initializeKeyManagerFactory() throws DrillException {
+ KeyManagerFactory kmf;
+ String keyStorePath = getKeyStorePath();
+ String keyStorePassword = getKeyStorePassword();
+ String keyStoreType = getKeyStoreType();
+ try {
+ if (keyStorePath.isEmpty()) {
+ throw new DrillException("No Keystore provided.");
+ }
+ KeyStore ks =
+ KeyStore.getInstance(!keyStoreType.isEmpty() ? keyStoreType : KeyStore.getDefaultType());
+ //initialize the key manager factory
+ // Will throw an exception if the file is not found/accessible.
+ InputStream ksStream = new FileInputStream(keyStorePath);
+ // A key password CANNOT be null or an empty string.
+ if (keyStorePassword.isEmpty()) {
+ throw new DrillException("The Keystore password cannot be empty.");
+ }
+ ks.load(ksStream, keyStorePassword.toCharArray());
+ // Empty Keystore. (Remarkably, it is possible to do this).
+ if (ks.size() == 0) {
+ throw new DrillException("The Keystore has no entries.");
+ }
+ kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(ks, getKeyPassword().toCharArray());
+
+ } catch (Exception e) {
+ throw new DrillException(
+ new StringBuilder()
+ .append("Exception while initializing the keystore: [")
+ .append(e.getMessage())
+ .append("]. ")
+ .toString());
+ }
+ return kmf;
+ }
+
+ public void initContext() throws DrillException {
+ if ((isWindows || isMacOs) && useSystemTrustStore()) {
+ initJDKSSLContext();
+ logger.debug("Initialized Windows/MacOs SSL context using JDK.");
+ } else {
+ initNettySslContext();
+ logger.debug("Initialized SSL context.");
+ }
+ return;
+ }
+
+ public SSLEngine createSSLEngine(BufferAllocator allocator, String peerHost, int peerPort) {
+ SSLEngine engine;
+ if ((isWindows || isMacOs) && useSystemTrustStore()) {
+ if (peerHost != null) {
+ engine = jdkSSlContext.createSSLEngine(peerHost, peerPort);
+ logger.debug("Initializing Windows/MacOs SSLEngine with hostname.");
+ } else {
+ engine = jdkSSlContext.createSSLEngine();
+ logger.debug("Initializing Windows/MacOs SSLEngine with no hostname.");
+ }
+ } else {
+ if (peerHost != null) {
+ engine = nettySslContext.newEngine(allocator.getAsByteBufAllocator(), peerHost, peerPort);
+ logger.debug("Initializing SSLEngine with hostname.");
+ } else {
+ engine = nettySslContext.newEngine(allocator.getAsByteBufAllocator());
+ logger.debug("Initializing SSLEngine with no hostname.");
+ }
+ }
+ return engine;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("SSL is ")
+ .append(isUserSslEnabled()?"":" not ")
+ .append("enabled.\n");
+ sb.append("HTTPS is ")
+ .append(isHttpsEnabled()?"":" not ")
+ .append("enabled.\n");
+ if(isUserSslEnabled() || isHttpsEnabled()) {
+ sb.append("SSL Configuration :")
+ .append("OS:").append(System.getProperty("os.name"))
+ .append("\n\tUsing system trust store: ").append(useSystemTrustStore())
+ .append("\n\tprotocol: ").append(getProtocol())
+ .append("\n\tkeyStoreType: ").append(getKeyStoreType())
+ .append("\n\tkeyStorePath: ").append(getKeyStorePath())
+ .append("\n\tkeyStorePassword: ").append(getPrintablePassword(getKeyStorePassword()))
+ .append("\n\tkeyPassword: ").append(getPrintablePassword(getKeyPassword()))
+ .append("\n\ttrustStoreType: ").append(getTrustStoreType())
+ .append("\n\ttrustStorePath: ").append(getTrustStorePath())
+ .append("\n\ttrustStorePassword: ").append(getPrintablePassword(getTrustStorePassword()))
+ .append("\n\thandshakeTimeout: ").append(getHandshakeTimeout())
+ .append("\n\tdisableHostVerification: ").append(disableHostVerification())
+ .append("\n\tdisableCertificateVerification: ").append(disableCertificateVerification())
+ ;
+ }
+ return sb.toString();
+ }
+
+ private String getPrintablePassword(String password) {
+ StringBuilder sb = new StringBuilder();
+ if(password == null || password.length()<2 ){
+ return password;
+ }
+ sb.append(password.charAt(0)).append("****").append(password.charAt(password.length()-1));
+ return sb.toString();
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigBuilder.java
new file mode 100644
index 000000000..0941960fa
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigBuilder.java
@@ -0,0 +1,94 @@
+/*
+ * 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.drill.exec.ssl;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.exceptions.DrillConfigurationException;
+import org.apache.drill.common.exceptions.DrillException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.ssl.SSLFactory;
+
+import java.util.Properties;
+
+
+public class SSLConfigBuilder {
+
+ private static final org.slf4j.Logger logger =
+ org.slf4j.LoggerFactory.getLogger(org.apache.drill.exec.ssl.SSLConfigBuilder.class);
+
+ private DrillConfig config = null;
+ private Configuration hadoopConfig = null;
+ private Properties properties;
+ private SSLFactory.Mode mode = SSLFactory.Mode.SERVER;
+ private boolean initializeSSLContext = false;
+ private boolean validateKeyStore = false;
+
+ public SSLConfigBuilder() {
+
+ }
+
+ public SSLConfig build() throws DrillException {
+ if (mode == SSLFactory.Mode.SERVER && config == null) {
+ throw new DrillConfigurationException(
+ "Cannot create SSL configuration from null Drill configuration.");
+ }
+ SSLConfig sslConfig;
+ if (mode == SSLFactory.Mode.SERVER) {
+ sslConfig = new SSLConfigServer(config, hadoopConfig);
+ } else {
+ sslConfig = new SSLConfigClient(properties);
+ }
+ if(validateKeyStore){
+ sslConfig.validateKeyStore();
+ }
+ if(initializeSSLContext){
+ sslConfig.initContext();
+ }
+ return sslConfig;
+ }
+
+ public SSLConfigBuilder config(DrillConfig config) {
+ this.config = config;
+ return this;
+ }
+
+ public SSLConfigBuilder hadoopConfig(Configuration hadoopConfig) {
+ this.hadoopConfig = hadoopConfig;
+ return this;
+ }
+
+ public SSLConfigBuilder properties(Properties props) {
+ this.properties = props;
+ return this;
+ }
+
+ public SSLConfigBuilder mode(SSLFactory.Mode mode) {
+ this.mode = mode;
+ return this;
+ }
+
+ public SSLConfigBuilder initializeSSLContext(boolean initializeSSLContext) {
+ this.initializeSSLContext = initializeSSLContext;
+ return this;
+ }
+
+ public SSLConfigBuilder validateKeyStore(boolean validateKeyStore) {
+ this.validateKeyStore = validateKeyStore;
+ return this;
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigClient.java
new file mode 100644
index 000000000..6a70d436a
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigClient.java
@@ -0,0 +1,273 @@
+/*
+ * 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.drill.exec.ssl;
+
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslProvider;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.config.DrillProperties;
+import org.apache.drill.common.exceptions.DrillException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.ssl.SSLFactory;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLParameters;
+import javax.net.ssl.TrustManagerFactory;
+import java.util.Properties;
+
+public class SSLConfigClient extends SSLConfig {
+
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SSLConfigClient.class);
+
+ private final Properties properties;
+ private final boolean userSslEnabled;
+ private final String trustStoreType;
+ private final String trustStorePath;
+ private final String trustStorePassword;
+ private final boolean disableHostVerification;
+ private final boolean disableCertificateVerification;
+ private final boolean useSystemTrustStore;
+ private final String protocol;
+ private final int handshakeTimeout;
+ private final String provider;
+
+ private final String emptyString = new String();
+
+ public SSLConfigClient(Properties properties) throws DrillException {
+ this.properties = properties;
+ userSslEnabled = getBooleanProperty(DrillProperties.ENABLE_TLS);
+ trustStoreType = getStringProperty(DrillProperties.TRUSTSTORE_TYPE, "JKS");
+ trustStorePath = getStringProperty(DrillProperties.TRUSTSTORE_PATH, "");
+ trustStorePassword = getStringProperty(DrillProperties.TRUSTSTORE_PASSWORD, "");
+ disableHostVerification = getBooleanProperty(DrillProperties.DISABLE_HOST_VERIFICATION);
+ disableCertificateVerification = getBooleanProperty(DrillProperties.DISABLE_CERT_VERIFICATION);
+ useSystemTrustStore = getBooleanProperty(DrillProperties.USE_SYSTEM_TRUSTSTORE);
+ protocol = getStringProperty(DrillProperties.TLS_PROTOCOL, DEFAULT_SSL_PROTOCOL);
+ int hsTimeout = getIntProperty(DrillProperties.TLS_HANDSHAKE_TIMEOUT, DEFAULT_SSL_HANDSHAKE_TIMEOUT_MS);
+ if (hsTimeout <= 0) {
+ hsTimeout = DEFAULT_SSL_HANDSHAKE_TIMEOUT_MS;
+ }
+ handshakeTimeout = hsTimeout;
+ provider = getStringProperty(DrillProperties.TLS_PROVIDER, DEFAULT_SSL_PROVIDER);
+ }
+
+ private boolean getBooleanProperty(String propName) {
+ return (properties != null) && (properties.containsKey(propName))
+ && (properties.getProperty(propName).compareToIgnoreCase("true") == 0);
+ }
+
+ private String getStringProperty(String name, String defaultValue) {
+ String value = "";
+ if ( (properties != null) && (properties.containsKey(name))) {
+ value = properties.getProperty(name);
+ }
+ if (value.isEmpty()) {
+ value = defaultValue;
+ }
+ value = value.trim();
+ return value;
+ }
+
+ private int getIntProperty(String name, int defaultValue) {
+ int value = defaultValue;
+ if ( (properties != null) && (properties.containsKey(name))) {
+ value = new Integer(properties.getProperty(name)).intValue();
+ }
+ return value;
+ }
+
+ public void validateKeyStore() throws DrillException {
+
+ }
+
+ @Override
+ public SslContext initNettySslContext() throws DrillException {
+ final SslContext sslCtx;
+
+ if (!userSslEnabled) {
+ return null;
+ }
+
+ TrustManagerFactory tmf;
+ try {
+ tmf = initializeTrustManagerFactory();
+ sslCtx = SslContextBuilder.forClient()
+ .sslProvider(getProvider())
+ .trustManager(tmf)
+ .protocols(protocol)
+ .build();
+ } catch (Exception e) {
+ // Catch any SSL initialization Exceptions here and abort.
+ throw new DrillException(new StringBuilder()
+ .append("SSL is enabled but cannot be initialized due to the following exception: ")
+ .append("[ ")
+ .append(e.getMessage())
+ .append("]. ")
+ .toString());
+ }
+ this.nettySslContext = sslCtx;
+ return sslCtx;
+ }
+
+ @Override
+ public SSLContext initJDKSSLContext() throws DrillException {
+ final SSLContext sslCtx;
+
+ if (!userSslEnabled) {
+ return null;
+ }
+
+ TrustManagerFactory tmf;
+ try {
+ tmf = initializeTrustManagerFactory();
+ sslCtx = SSLContext.getInstance(protocol);
+ sslCtx.init(null, tmf.getTrustManagers(), null);
+ } catch (Exception e) {
+ // Catch any SSL initialization Exceptions here and abort.
+ throw new DrillException(new StringBuilder()
+ .append("SSL is enabled but cannot be initialized due to the following exception: ")
+ .append("[ ")
+ .append(e.getMessage())
+ .append("]. ")
+ .toString());
+ }
+ this.jdkSSlContext = sslCtx;
+ return sslCtx;
+ }
+
+ @Override
+ public SSLEngine createSSLEngine(BufferAllocator allocator, String peerHost, int peerPort) {
+ SSLEngine engine = super.createSSLEngine(allocator, peerHost, peerPort);
+
+ if (!this.disableHostVerification()) {
+ SSLParameters sslParameters = engine.getSSLParameters();
+ // only available since Java 7
+ sslParameters.setEndpointIdentificationAlgorithm("HTTPS");
+ engine.setSSLParameters(sslParameters);
+ }
+
+ engine.setUseClientMode(true);
+
+ try {
+ engine.setEnableSessionCreation(true);
+ } catch (Exception e) {
+ // Openssl implementation may throw this.
+ logger.debug("Session creation not enabled. Exception: {}", e.getMessage());
+ }
+
+ return engine;
+ }
+
+ @Override
+ public boolean isUserSslEnabled() {
+ return userSslEnabled;
+ }
+
+ @Override
+ public boolean isHttpsEnabled() {
+ return false;
+ }
+
+ @Override
+ public String getKeyStoreType() {
+ return emptyString;
+ }
+
+ @Override
+ public String getKeyStorePath() {
+ return emptyString;
+ }
+
+ @Override
+ public String getKeyStorePassword() {
+ return emptyString;
+ }
+
+ @Override
+ public String getKeyPassword() {
+ return emptyString;
+ }
+
+ @Override
+ public String getTrustStoreType() {
+ return trustStoreType;
+ }
+
+ @Override
+ public boolean hasTrustStorePath() {
+ return !trustStorePath.isEmpty();
+ }
+
+ @Override
+ public String getTrustStorePath() {
+ return trustStorePath;
+ }
+
+ @Override
+ public boolean hasTrustStorePassword() {
+ return !trustStorePassword.isEmpty();
+ }
+
+ @Override
+ public String getTrustStorePassword() {
+ return trustStorePassword;
+ }
+
+ @Override
+ public String getProtocol() {
+ return protocol;
+ }
+
+ @Override
+ public SslProvider getProvider() {
+ return provider.equalsIgnoreCase("JDK") ? SslProvider.JDK : SslProvider.OPENSSL;
+ }
+
+ @Override
+ public int getHandshakeTimeout() {
+ return handshakeTimeout;
+ }
+
+ @Override
+ public SSLFactory.Mode getMode() {
+ return SSLFactory.Mode.CLIENT;
+ }
+
+ @Override
+ public boolean disableHostVerification() {
+ return disableHostVerification;
+ }
+
+ @Override
+ public boolean disableCertificateVerification() {
+ return disableCertificateVerification;
+ }
+
+ @Override
+ public boolean useSystemTrustStore() {
+ return useSystemTrustStore;
+ }
+
+ public boolean isSslValid() {
+ return true;
+ }
+
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigServer.java
new file mode 100644
index 000000000..da31d2e7f
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ssl/SSLConfigServer.java
@@ -0,0 +1,330 @@
+/*
+ * 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.drill.exec.ssl;
+
+import com.google.common.base.Preconditions;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslProvider;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.exceptions.DrillException;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.ssl.SSLFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.TrustManagerFactory;
+import java.text.MessageFormat;
+
+public class SSLConfigServer extends SSLConfig {
+
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SSLConfigServer.class);
+
+ private final DrillConfig config;
+ private final Configuration hadoopConfig;
+ private final boolean userSslEnabled;
+ private final boolean httpsEnabled;
+ private final String keyStoreType;
+ private final String keyStorePath;
+ private final String keyStorePassword;
+ private final String keyPassword;
+ private final String trustStoreType;
+ private final String trustStorePath;
+ private final String trustStorePassword;
+ private final String protocol;
+ private final String provider;
+
+ public SSLConfigServer(DrillConfig config, Configuration hadoopConfig) throws DrillException {
+ this.config = config;
+ SSLFactory.Mode mode = SSLFactory.Mode.SERVER;
+ httpsEnabled =
+ config.hasPath(ExecConstants.HTTP_ENABLE_SSL) && config.getBoolean(ExecConstants.HTTP_ENABLE_SSL);
+ // For testing we will mock up a hadoop configuration, however for regular use, we find the actual hadoop config.
+ boolean enableHadoopConfig = config.getBoolean(ExecConstants.SSL_USE_HADOOP_CONF);
+ if (enableHadoopConfig) {
+ if (hadoopConfig == null) {
+ this.hadoopConfig = new Configuration(); // get hadoop configuration
+ } else {
+ this.hadoopConfig = hadoopConfig;
+ }
+ String hadoopSSLConfigFile =
+ this.hadoopConfig.get(resolveHadoopPropertyName(HADOOP_SSL_CONF_TPL_KEY, getMode()));
+ logger.debug("Using Hadoop configuration for SSL");
+ logger.debug("Hadoop SSL configuration file: {}", hadoopSSLConfigFile);
+ this.hadoopConfig.addResource(hadoopSSLConfigFile);
+ } else {
+ this.hadoopConfig = null;
+ }
+ userSslEnabled =
+ config.hasPath(ExecConstants.USER_SSL_ENABLED) && config.getBoolean(ExecConstants.USER_SSL_ENABLED);
+ trustStoreType = getConfigParam(ExecConstants.SSL_TRUSTSTORE_TYPE,
+ resolveHadoopPropertyName(HADOOP_SSL_TRUSTSTORE_TYPE_TPL_KEY, mode));
+ trustStorePath = getConfigParam(ExecConstants.SSL_TRUSTSTORE_PATH,
+ resolveHadoopPropertyName(HADOOP_SSL_TRUSTSTORE_LOCATION_TPL_KEY, mode));
+ trustStorePassword = getConfigParam(ExecConstants.SSL_TRUSTSTORE_PASSWORD,
+ resolveHadoopPropertyName(HADOOP_SSL_TRUSTSTORE_PASSWORD_TPL_KEY, mode));
+ keyStoreType = getConfigParam(ExecConstants.SSL_KEYSTORE_TYPE,
+ resolveHadoopPropertyName(HADOOP_SSL_KEYSTORE_TYPE_TPL_KEY, mode));
+ keyStorePath = getConfigParam(ExecConstants.SSL_KEYSTORE_PATH,
+ resolveHadoopPropertyName(HADOOP_SSL_KEYSTORE_LOCATION_TPL_KEY, mode));
+ keyStorePassword = getConfigParam(ExecConstants.SSL_KEYSTORE_PASSWORD,
+ resolveHadoopPropertyName(HADOOP_SSL_KEYSTORE_PASSWORD_TPL_KEY, mode));
+ // if no keypassword specified, use keystore password
+ String keyPass = getConfigParam(ExecConstants.SSL_KEY_PASSWORD,
+ resolveHadoopPropertyName(HADOOP_SSL_KEYSTORE_KEYPASSWORD_TPL_KEY, mode));
+ keyPassword = keyPass.isEmpty() ? keyStorePassword : keyPass;
+ protocol = getConfigParamWithDefault(ExecConstants.SSL_PROTOCOL, DEFAULT_SSL_PROTOCOL);
+ provider = getConfigParamWithDefault(ExecConstants.SSL_PROVIDER, DEFAULT_SSL_PROVIDER);
+ }
+
+ public void validateKeyStore() throws DrillException {
+ //HTTPS validates the keystore is not empty. User Server SSL context initialization also validates keystore, but
+ // much more strictly. User Client context initialization does not validate keystore.
+ /*If keystorePath or keystorePassword is provided in the configuration file use that*/
+ if ((isUserSslEnabled() || isHttpsEnabled())) {
+ if (!keyStorePath.isEmpty() || !keyStorePassword.isEmpty()) {
+ if (keyStorePath.isEmpty()) {
+ throw new DrillException(
+ " *.ssl.keyStorePath in the configuration file is empty, but *.ssl.keyStorePassword is set");
+ } else if (keyStorePassword.isEmpty()) {
+ throw new DrillException(
+ " *.ssl.keyStorePassword in the configuration file is empty, but *.ssl.keyStorePath is set ");
+ }
+ }
+ }
+ }
+
+ @Override
+ public SslContext initNettySslContext() throws DrillException {
+ final SslContext sslCtx;
+
+ if (!userSslEnabled) {
+ return null;
+ }
+
+ KeyManagerFactory kmf;
+ TrustManagerFactory tmf;
+ try {
+ if (keyStorePath.isEmpty()) {
+ throw new DrillException("No Keystore provided.");
+ }
+ kmf = initializeKeyManagerFactory();
+ tmf = initializeTrustManagerFactory();
+ sslCtx = SslContextBuilder.forServer(kmf)
+ .trustManager(tmf)
+ .protocols(protocol)
+ .sslProvider(getProvider())
+ .build(); // Will throw an exception if the key password is not correct
+ } catch (Exception e) {
+ // Catch any SSL initialization Exceptions here and abort.
+ throw new DrillException(new StringBuilder()
+ .append("SSL is enabled but cannot be initialized - ")
+ .append("[ ")
+ .append(e.getMessage())
+ .append("]. ")
+ .toString());
+ }
+ this.nettySslContext = sslCtx;
+ return sslCtx;
+ }
+
+ @Override
+ public SSLContext initJDKSSLContext() throws DrillException {
+ final SSLContext sslCtx;
+
+ if (!userSslEnabled) {
+ return null;
+ }
+
+ KeyManagerFactory kmf;
+ TrustManagerFactory tmf;
+ try {
+ if (keyStorePath.isEmpty()) {
+ throw new DrillException("No Keystore provided.");
+ }
+ kmf = initializeKeyManagerFactory();
+ tmf = initializeTrustManagerFactory();
+ sslCtx = SSLContext.getInstance(protocol);
+ sslCtx.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
+ } catch (Exception e) {
+ // Catch any SSL initialization Exceptions here and abort.
+ throw new DrillException(
+ new StringBuilder().append("SSL is enabled but cannot be initialized - ")
+ .append("[ ")
+ .append(e.getMessage())
+ .append("]. ")
+ .toString());
+ }
+ this.jdkSSlContext = sslCtx;
+ return sslCtx;
+ }
+
+ @Override
+ public SSLEngine createSSLEngine(BufferAllocator allocator, String peerHost, int peerPort) {
+ SSLEngine engine = super.createSSLEngine(allocator, peerHost, peerPort);
+
+ engine.setUseClientMode(false);
+
+ // No need for client side authentication (HTTPS like behaviour)
+ engine.setNeedClientAuth(false);
+
+ try {
+ engine.setEnableSessionCreation(true);
+ } catch (Exception e) {
+ // Openssl implementation may throw this.
+ logger.debug("Session creation not enabled. Exception: {}", e.getMessage());
+ }
+
+ return engine;
+ }
+
+ private String getConfigParam(String name, String hadoopName) {
+ String value = "";
+ if (hadoopConfig != null) {
+ value = getHadoopConfigParam(hadoopName);
+ }
+ if (value.isEmpty() && config.hasPath(name)) {
+ value = config.getString(name);
+ }
+ value = value.trim();
+ return value;
+ }
+
+ private String getHadoopConfigParam(String name) {
+ Preconditions.checkArgument(this.hadoopConfig != null);
+ String value = hadoopConfig.get(name, "");
+ value = value.trim();
+ return value;
+ }
+
+ private String getConfigParamWithDefault(String name, String defaultValue) {
+ String value = "";
+ if (config.hasPath(name)) {
+ value = config.getString(name);
+ }
+ if (value.isEmpty()) {
+ value = defaultValue;
+ }
+ value = value.trim();
+ return value;
+ }
+
+ private String resolveHadoopPropertyName(String nameTemplate, SSLFactory.Mode mode) {
+ return MessageFormat.format(nameTemplate, mode.toString().toLowerCase());
+ }
+
+
+
+ @Override
+ public boolean isUserSslEnabled() {
+ return userSslEnabled;
+ }
+
+ @Override
+ public boolean isHttpsEnabled() {
+ return httpsEnabled;
+ }
+
+ @Override
+ public String getKeyStoreType() {
+ return keyStoreType;
+ }
+
+ @Override
+ public String getKeyStorePath() {
+ return keyStorePath;
+ }
+
+ @Override
+ public String getKeyStorePassword() {
+ return keyStorePassword;
+ }
+
+ @Override
+ public String getKeyPassword() {
+ return keyPassword;
+ }
+
+ @Override
+ public String getTrustStoreType() {
+ return trustStoreType;
+ }
+
+ @Override
+ public boolean hasTrustStorePath() {
+ return !trustStorePath.isEmpty();
+ }
+
+ @Override
+ public String getTrustStorePath() {
+ return trustStorePath;
+ }
+
+ @Override
+ public boolean hasTrustStorePassword() {
+ return !trustStorePassword.isEmpty();
+ }
+
+ @Override
+ public String getTrustStorePassword() {
+ return trustStorePassword;
+ }
+
+ @Override
+ public String getProtocol() {
+ return protocol;
+ }
+
+ @Override
+ public SslProvider getProvider() {
+ return provider.equalsIgnoreCase("JDK") ? SslProvider.JDK : SslProvider.OPENSSL;
+ }
+
+ @Override
+ public int getHandshakeTimeout() {
+ return 0;
+ }
+
+ @Override
+ public SSLFactory.Mode getMode() {
+ return SSLFactory.Mode.SERVER;
+ }
+
+ @Override
+ public boolean disableHostVerification() {
+ return false;
+ }
+
+ @Override
+ public boolean disableCertificateVerification() {
+ return false;
+ }
+
+ @Override
+ public boolean useSystemTrustStore() {
+ return false; // Client only, notsupported by the server
+ }
+
+ public boolean isSslValid() {
+ return !keyStorePath.isEmpty() && !keyStorePassword.isEmpty();
+ }
+
+}