You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by su...@apache.org on 2017/02/25 07:18:11 UTC
[18/29] drill git commit: DRILL-4280: CORE (user to bit
authentication, Java)
DRILL-4280: CORE (user to bit authentication, Java)
+ Add logic for authentication in UserClient and UserServer with
backward compatibility in both directions
+ Add abstract extension to ServerConnection and ClientConnection
+ Add concrete extensions to abstract connections:
BitToUserConnection and UserToBitConnection
+ Add ConnectionConfig interface with abstract and concrete
implementations to encapsulate configuration for server-side
connections
+ Encapsulate all requests handled by UserServer in
UserServerRequestHandler
+ Clear UserSession when connection is closed either by user or
bit
+ Add DrillProperties to encapsulate all connection properties
used during connection time
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/b6e59ec3
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/b6e59ec3
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/b6e59ec3
Branch: refs/heads/master
Commit: b6e59ec350c7cc7f467896abba4476cfcfc9077e
Parents: 8a732c0
Author: Sudheesh Katkam <su...@apache.org>
Authored: Wed Jan 25 19:02:45 2017 -0800
Committer: Sudheesh Katkam <su...@apache.org>
Committed: Fri Feb 24 19:01:42 2017 -0800
----------------------------------------------------------------------
.../drill/common/config/DrillProperties.java | 186 ++++++++++++
.../apache/drill/exec/client/DrillClient.java | 94 ++----
.../client/InvalidConnectionInfoException.java | 30 ++
.../exec/rpc/AbstractClientConnection.java | 63 ++++
.../exec/rpc/AbstractConnectionConfig.java | 49 +++
.../exec/rpc/AbstractServerConnection.java | 124 ++++++++
.../apache/drill/exec/rpc/ConnectionConfig.java | 34 +++
.../drill/exec/rpc/user/QueryResultHandler.java | 16 +-
.../apache/drill/exec/rpc/user/UserClient.java | 281 ++++++++++++++++--
.../exec/rpc/user/UserConnectionConfig.java | 74 +++++
.../drill/exec/rpc/user/UserRpcConfig.java | 10 +-
.../apache/drill/exec/rpc/user/UserServer.java | 295 ++++++++-----------
.../exec/rpc/user/UserServerRequestHandler.java | 143 +++++++++
.../apache/drill/exec/rpc/user/UserSession.java | 42 +--
.../drill/exec/service/ServiceEngine.java | 28 +-
.../drill/jdbc/impl/DrillConnectionImpl.java | 2 +-
16 files changed, 1141 insertions(+), 330 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/common/src/main/java/org/apache/drill/common/config/DrillProperties.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/drill/common/config/DrillProperties.java b/common/src/main/java/org/apache/drill/common/config/DrillProperties.java
new file mode 100644
index 0000000..ba63349
--- /dev/null
+++ b/common/src/main/java/org/apache/drill/common/config/DrillProperties.java
@@ -0,0 +1,186 @@
+/*
+ * 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.common.config;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.drill.exec.proto.UserProtos.Property;
+import org.apache.drill.exec.proto.UserProtos.UserProperties;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+public final class DrillProperties extends Properties {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillProperties.class);
+
+ // PROPERTY KEYS
+ // definitions should be in lowercase
+
+ public static final String ZOOKEEPER_CONNECTION = "zk";
+
+ public static final String DRILLBIT_CONNECTION = "drillbit";
+
+ // "tries" is max number of unique drillbits to try connecting
+ // until successfully connected to one of them
+ public static final String TRIES = "tries";
+
+ public static final String SCHEMA = "schema";
+
+ public static final String USER = "user";
+
+ public static final String PASSWORD = "password";
+
+ public static final String IMPERSONATION_TARGET = "impersonation_target";
+
+ public static final String AUTH_MECHANISM = "auth";
+
+ public static final String SERVICE_PRINCIPAL = "principal";
+
+ public static final String SERVICE_NAME = "service_name";
+
+ public static final String SERVICE_HOST = "service_host";
+
+ public static final String REALM = "realm";
+
+ public static final String KEYTAB = "keytab";
+
+ // for subject that has pre-authenticated to KDC (AS) i.e. required credentials are populated in
+ // Subject's credentials set
+ public static final String KERBEROS_FROM_SUBJECT = "from_subject";
+
+ // Although all properties from the application are sent to the server (from the client), the following
+ // sets of properties are used by the client and server respectively. These are reserved words.
+
+ public static final ImmutableSet<String> ALLOWED_BY_CLIENT =
+ ImmutableSet.of(
+ ZOOKEEPER_CONNECTION, DRILLBIT_CONNECTION, TRIES,
+ SCHEMA,
+ USER, PASSWORD, IMPERSONATION_TARGET, AUTH_MECHANISM,
+ SERVICE_PRINCIPAL, SERVICE_NAME, SERVICE_HOST, REALM, KEYTAB, KERBEROS_FROM_SUBJECT
+ );
+
+ public static final ImmutableSet<String> ACCEPTED_BY_SERVER = ImmutableSet.of(
+ USER /** deprecated */, PASSWORD /** deprecated */,
+ SCHEMA,
+ IMPERSONATION_TARGET
+ );
+
+ private DrillProperties() {
+ }
+
+ @Override
+ public Object setProperty(final String key, final String value) {
+ return super.setProperty(key.toLowerCase(), value);
+ }
+
+ @Override
+ public String getProperty(final String key) {
+ return super.getProperty(key.toLowerCase());
+ }
+
+ @Override
+ public String getProperty(final String key, final String defaultValue) {
+ return super.getProperty(key.toLowerCase(), defaultValue);
+ }
+
+ public void merge(final Properties overrides) {
+ if (overrides == null) {
+ return;
+ }
+ for (final String key : overrides.stringPropertyNames()) {
+ setProperty(key.toLowerCase(), overrides.getProperty(key));
+ }
+ }
+
+ /**
+ * Returns a map of keys and values in this property list where the key and its corresponding value are strings,
+ * including distinct keys in the default property list if a key of the same name has not already been found from
+ * the main properties list. Properties whose key or value is not of type <tt>String</tt> are omitted.
+ * <p>
+ * The returned map is not backed by the <tt>Properties</tt> object. Changes to this <tt>Properties</tt> are not
+ * reflected in the map, or vice versa.
+ *
+ * @return a map of keys and values in this property list where the key and its corresponding value are strings,
+ * including the keys in the default property list.
+ */
+ public Map<String, String> stringPropertiesAsMap() {
+ final Map<String, String> map = new HashMap<>();
+ for (final String property : stringPropertyNames()) {
+ map.put(property, getProperty(property));
+ }
+ return map;
+ }
+
+ /**
+ * Serializes properties into a protobuf message.
+ *
+ * @return the serialized properties
+ */
+ public UserProperties serializeForServer() {
+ final UserProperties.Builder propsBuilder = UserProperties.newBuilder();
+ for (final String key : stringPropertyNames()) {
+ propsBuilder.addProperties(Property.newBuilder()
+ .setKey(key)
+ .setValue(getProperty(key))
+ .build());
+ }
+ return propsBuilder.build();
+ }
+
+ /**
+ * Deserializes the given properties into DrillProperties.
+ *
+ * @param userProperties serialized user properties
+ * @param addOnlyKnownServerProperties add only properties known by server
+ * @return params
+ */
+ public static DrillProperties createFromProperties(final UserProperties userProperties,
+ final boolean addOnlyKnownServerProperties) {
+ final DrillProperties properties = new DrillProperties();
+ for (final Property property : userProperties.getPropertiesList()) {
+ final String key = property.getKey().toLowerCase();
+ if (!addOnlyKnownServerProperties || ACCEPTED_BY_SERVER.contains(key)) {
+ properties.setProperty(key, property.getValue());
+ } else {
+ logger.trace("Server does not recognize property: {}", key);
+ }
+ }
+ return properties;
+ }
+
+ /**
+ * Returns a new instance of DrillProperties from the given properties.
+ *
+ * @param properties user properties
+ * @return params
+ */
+ public static DrillProperties createFromProperties(final Properties properties) {
+ final DrillProperties drillProperties = new DrillProperties();
+ if (properties != null) {
+ for (final String key : properties.stringPropertyNames()) {
+ final String lowerCaseKey = key.toLowerCase();
+ drillProperties.setProperty(lowerCaseKey, properties.getProperty(key));
+ }
+ }
+ return drillProperties;
+ }
+
+ public static DrillProperties createEmpty() {
+ return new DrillProperties();
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
----------------------------------------------------------------------
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 dee0772..508c722 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
@@ -23,6 +23,9 @@ import static com.google.common.base.Preconditions.checkState;
import static org.apache.drill.exec.proto.UserProtos.QueryResultsMode.STREAM_FULL;
import static org.apache.drill.exec.proto.UserProtos.RunQuery.newBuilder;
+import com.google.common.base.Strings;
+import io.netty.channel.EventLoopGroup;
+
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
@@ -36,6 +39,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.drill.common.DrillAutoCloseables;
+import org.apache.drill.common.config.DrillProperties;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.ExecConstants;
@@ -65,22 +69,18 @@ import org.apache.drill.exec.proto.UserProtos.GetTablesReq;
import org.apache.drill.exec.proto.UserProtos.GetTablesResp;
import org.apache.drill.exec.proto.UserProtos.LikeFilter;
import org.apache.drill.exec.proto.UserProtos.PreparedStatementHandle;
-import org.apache.drill.exec.proto.UserProtos.Property;
import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
import org.apache.drill.exec.proto.UserProtos.RpcEndpointInfos;
import org.apache.drill.exec.proto.UserProtos.RpcType;
import org.apache.drill.exec.proto.UserProtos.RunQuery;
-import org.apache.drill.exec.proto.UserProtos.UserProperties;
import org.apache.drill.exec.proto.helper.QueryIdHelper;
-import org.apache.drill.exec.rpc.BasicClientWithConnection.ServerConnection;
import org.apache.drill.exec.rpc.ChannelClosedException;
import org.apache.drill.exec.rpc.ConnectionThrottle;
import org.apache.drill.exec.rpc.DrillRpcFuture;
import org.apache.drill.exec.rpc.NamedThreadFactory;
-import org.apache.drill.exec.rpc.RpcConnectionHandler;
import org.apache.drill.exec.rpc.RpcException;
-import org.apache.drill.exec.rpc.InvalidConnectionInfoException;
import org.apache.drill.exec.rpc.TransportCheck;
+import org.apache.drill.exec.rpc.NonTransientRpcException;
import org.apache.drill.exec.rpc.user.QueryDataBatch;
import org.apache.drill.exec.rpc.user.UserClient;
import org.apache.drill.exec.rpc.user.UserResultsListener;
@@ -89,12 +89,8 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Strings;
-import com.google.common.util.concurrent.AbstractCheckedFuture;
-import com.google.common.util.concurrent.SettableFuture;
-import io.netty.buffer.DrillBuf;
-import io.netty.channel.EventLoopGroup;
+import com.google.common.util.concurrent.SettableFuture;
/**
* Thin wrapper around a UserClient that handles connect/close and transforms
@@ -108,7 +104,7 @@ public class DrillClient implements Closeable, ConnectionThrottle {
private static final ObjectMapper objectMapper = new ObjectMapper();
private final DrillConfig config;
private UserClient client;
- private UserProperties props = null;
+ private DrillProperties properties;
private volatile ClusterCoordinator clusterCoordinator;
private volatile boolean connected = false;
private final BufferAllocator allocator;
@@ -312,12 +308,13 @@ public class DrillClient implements Closeable, ConnectionThrottle {
if (connected) {
return;
}
+ properties = DrillProperties.createFromProperties(props);
final List<DrillbitEndpoint> endpoints = new ArrayList<>();
if (isDirectConnection) {
// Populate the endpoints list with all the drillbit information provided in the connection string
- endpoints.addAll(parseAndVerifyEndpoints(props.getProperty("drillbit"),
+ endpoints.addAll(parseAndVerifyEndpoints(properties.getProperty(DrillProperties.DRILLBIT_CONNECTION),
config.getString(ExecConstants.INITIAL_USER_PORT)));
} else {
if (ownsZkConnection) {
@@ -330,21 +327,12 @@ public class DrillClient implements Closeable, ConnectionThrottle {
}
endpoints.addAll(clusterCoordinator.getAvailableEndpoints());
// Make sure we have at least one endpoint in the list
- checkState(!endpoints.isEmpty(), "No active Drillbit endpoint found from ZooKeeper");
+ checkState(!endpoints.isEmpty(), "No active Drillbit endpoint found from ZooKeeper. Check connection parameters?");
}
// shuffle the collection then get the first endpoint
Collections.shuffle(endpoints);
- if (props != null) {
- final UserProperties.Builder upBuilder = UserProperties.newBuilder();
- for (final String key : props.stringPropertyNames()) {
- upBuilder.addProperties(Property.newBuilder().setKey(key).setValue(props.getProperty(key)));
- }
-
- this.props = upBuilder.build();
- }
-
eventLoopGroup = createEventLoop(config.getInt(ExecConstants.CLIENT_RPC_THREADS), "Client-");
executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
@@ -358,9 +346,7 @@ public class DrillClient implements Closeable, ConnectionThrottle {
}
};
- // "tries" is max number of unique drillbit to try connecting until successfully connected to one of them
- final String connectTriesConf = (props != null) ? props.getProperty("tries", "5") : "5";
-
+ final String connectTriesConf = properties.getProperty(DrillProperties.TRIES, "5");
int connectTriesVal;
try {
connectTriesVal = Math.min(endpoints.size(), Integer.parseInt(connectTriesConf));
@@ -381,12 +367,16 @@ public class DrillClient implements Closeable, ConnectionThrottle {
endpoint = endpoints.get(triedEndpointIndex);
logger.debug("Connecting to server {}:{}", endpoint.getAddress(), endpoint.getUserPort());
+ if (!properties.containsKey(DrillProperties.SERVICE_HOST)) {
+ properties.setProperty(DrillProperties.SERVICE_HOST, endpoint.getAddress());
+ }
+
try {
connect(endpoint);
connected = true;
logger.info("Successfully connected to server {}:{}", endpoint.getAddress(), endpoint.getUserPort());
break;
- } catch (InvalidConnectionInfoException ex) {
+ } catch (NonTransientRpcException ex) {
logger.error("Connection to {}:{} failed with error {}. Not retrying anymore", endpoint.getAddress(),
endpoint.getUserPort(), ex.getMessage());
throw ex;
@@ -435,9 +425,7 @@ public class DrillClient implements Closeable, ConnectionThrottle {
}
private void connect(DrillbitEndpoint endpoint) throws RpcException {
- final FutureHandler f = new FutureHandler();
- client.connect(f, endpoint, props, getUserCredentials());
- f.checkedGet();
+ client.connect(endpoint, properties, getUserCredentials());
}
public BufferAllocator getAllocator() {
@@ -568,19 +556,13 @@ public class DrillClient implements Closeable, ConnectionThrottle {
* Helper method to generate the UserCredentials message from the properties.
*/
private UserBitShared.UserCredentials getUserCredentials() {
- // If username is not propagated as one of the properties
- String userName = "anonymous";
-
- if (props != null) {
- for (Property property: props.getPropertiesList()) {
- if (property.getKey().equalsIgnoreCase("user") && !Strings.isNullOrEmpty(property.getValue())) {
- userName = property.getValue();
- break;
- }
- }
+ String userName = properties.getProperty(DrillProperties.USER);
+ if (Strings.isNullOrEmpty(userName)) {
+ userName = "anonymous"; // if username is not propagated as one of the properties
}
-
- return UserBitShared.UserCredentials.newBuilder().setUserName(userName).build();
+ return UserBitShared.UserCredentials.newBuilder()
+ .setUserName(userName)
+ .build();
}
public DrillRpcFuture<Ack> cancelQuery(QueryId id) {
@@ -827,34 +809,4 @@ public class DrillClient implements Closeable, ConnectionThrottle {
}
}
}
-
- private class FutureHandler extends AbstractCheckedFuture<Void, RpcException> implements RpcConnectionHandler<ServerConnection>, DrillRpcFuture<Void>{
- protected FutureHandler() {
- super( SettableFuture.<Void>create());
- }
-
- @Override
- public void connectionSucceeded(ServerConnection connection) {
- getInner().set(null);
- }
-
- @Override
- public void connectionFailed(FailureType type, Throwable t) {
- getInner().setException(new RpcException(String.format("%s : %s", type.name(), t.getMessage()), t));
- }
-
- private SettableFuture<Void> getInner() {
- return (SettableFuture<Void>) delegate();
- }
-
- @Override
- protected RpcException mapException(Exception e) {
- return RpcException.mapException(e);
- }
-
- @Override
- public DrillBuf getBuffer() {
- return null;
- }
- }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/client/InvalidConnectionInfoException.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/InvalidConnectionInfoException.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/InvalidConnectionInfoException.java
new file mode 100644
index 0000000..19e72ff
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/InvalidConnectionInfoException.java
@@ -0,0 +1,30 @@
+/*
+ * 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.client;
+
+import org.apache.drill.exec.rpc.NonTransientRpcException;
+
+/**
+ * Exception for malformed connection string from client
+ */
+public class InvalidConnectionInfoException extends NonTransientRpcException {
+
+ public InvalidConnectionInfoException(String message) {
+ super(message);
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractClientConnection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractClientConnection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractClientConnection.java
new file mode 100644
index 0000000..055ea59
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractClientConnection.java
@@ -0,0 +1,63 @@
+/*
+ * 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.rpc;
+
+import io.netty.channel.socket.SocketChannel;
+import org.slf4j.Logger;
+
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public abstract class AbstractClientConnection extends AbstractRemoteConnection implements ClientConnection {
+// private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(AbstractClientConnection.class);
+
+ private SaslClient saslClient;
+
+ public AbstractClientConnection(SocketChannel channel, String name) {
+ super(channel, name);
+ }
+
+ protected abstract Logger getLogger();
+
+ @Override
+ public void setSaslClient(final SaslClient saslClient) {
+ checkState(this.saslClient == null);
+ this.saslClient = saslClient;
+ }
+
+ @Override
+ public SaslClient getSaslClient() {
+ checkState(this.saslClient != null);
+ return saslClient;
+ }
+
+ @Override
+ public void close() {
+ try {
+ if (saslClient != null) {
+ saslClient.dispose();
+ saslClient = null;
+ }
+ } catch (final SaslException e) {
+ getLogger().warn("Unclean disposal", e);
+ }
+ super.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractConnectionConfig.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractConnectionConfig.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractConnectionConfig.java
new file mode 100644
index 0000000..fb815ab
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractConnectionConfig.java
@@ -0,0 +1,49 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.rpc;
+
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.rpc.security.AuthenticatorProvider;
+import org.apache.drill.exec.server.BootStrapContext;
+
+public abstract class AbstractConnectionConfig implements ConnectionConfig {
+// private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(AbstractConnectionConfig.class);
+
+ private final BufferAllocator allocator;
+ private final BootStrapContext context;
+
+ protected AbstractConnectionConfig(BufferAllocator allocator, BootStrapContext context) {
+ this.allocator = allocator;
+ this.context = context;
+ }
+
+ @Override
+ public BootStrapContext getBootstrapContext() {
+ return context;
+ }
+
+ @Override
+ public BufferAllocator getAllocator() {
+ return allocator;
+ }
+
+ @Override
+ public AuthenticatorProvider getAuthProvider() {
+ return context.getAuthProvider();
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractServerConnection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractServerConnection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractServerConnection.java
new file mode 100644
index 0000000..72af064
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractServerConnection.java
@@ -0,0 +1,124 @@
+/*
+ * 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.rpc;
+
+import io.netty.channel.socket.SocketChannel;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.hadoop.security.HadoopKerberosName;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+
+import javax.security.auth.login.LoginException;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import java.io.IOException;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+public abstract class AbstractServerConnection<S extends ServerConnection<S>>
+ extends AbstractRemoteConnection
+ implements ServerConnection<S> {
+
+ private final ConnectionConfig config;
+
+ private RequestHandler<S> currentHandler;
+ private SaslServer saslServer;
+
+ public AbstractServerConnection(SocketChannel channel, String name, ConnectionConfig config,
+ RequestHandler<S> handler) {
+ super(channel, name);
+ this.config = config;
+ this.currentHandler = handler;
+ }
+
+ public AbstractServerConnection(SocketChannel channel, ConnectionConfig config,
+ RequestHandler<S> handler) {
+ this(channel, config.getName(), config, handler);
+ }
+
+ @Override
+ public BufferAllocator getAllocator() {
+ return config.getAllocator();
+ }
+
+ protected abstract Logger getLogger();
+
+ @Override
+ public void initSaslServer(String mechanismName) throws SaslException {
+ checkState(saslServer == null);
+ try {
+ this.saslServer = config.getAuthProvider()
+ .getAuthenticatorFactory(mechanismName)
+ .createSaslServer(UserGroupInformation.getLoginUser(), null
+ /** properties; default QOP is auth */);
+ } catch (final IOException e) {
+ getLogger().debug("Login failed.", e);
+ final Throwable cause = e.getCause();
+ if (cause instanceof LoginException) {
+ throw new SaslException("Failed to login.", cause);
+ }
+ throw new SaslException("Unexpected failure trying to login.", cause);
+ }
+ if (saslServer == null) {
+ throw new SaslException("Server could not initiate authentication. Insufficient parameters?");
+ }
+ }
+
+ @Override
+ public SaslServer getSaslServer() {
+ checkState(saslServer != null);
+ return saslServer;
+ }
+
+ @Override
+ public void finalizeSession() throws IOException {
+ final String authorizationID = getSaslServer().getAuthorizationID();
+ final String remoteShortName = new HadoopKerberosName(authorizationID).getShortName();
+ final String localShortName = UserGroupInformation.getLoginUser().getShortUserName();
+ if (!localShortName.equals(remoteShortName)) {
+ throw new SaslException(String.format("'primary' part of remote drillbit's service principal " +
+ "does not match with this drillbit's. Expected: '%s' Actual: '%s'", localShortName, remoteShortName));
+ }
+ getLogger().debug("Authenticated connection for {}", authorizationID);
+ }
+
+ @Override
+ public RequestHandler<S> getCurrentHandler() {
+ return currentHandler;
+ }
+
+ @Override
+ public void changeHandlerTo(final RequestHandler<S> handler) {
+ checkNotNull(handler);
+ this.currentHandler = handler;
+ }
+
+ @Override
+ public void close() {
+ try {
+ if (saslServer != null) {
+ saslServer.dispose();
+ saslServer = null;
+ }
+ } catch (final SaslException e) {
+ getLogger().warn("Unclean disposal.", e);
+ }
+ super.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ConnectionConfig.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ConnectionConfig.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ConnectionConfig.java
new file mode 100644
index 0000000..706b088
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ConnectionConfig.java
@@ -0,0 +1,34 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.rpc;
+
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.rpc.security.AuthenticatorProvider;
+import org.apache.drill.exec.server.BootStrapContext;
+
+public interface ConnectionConfig {
+
+ String getName();
+
+ BootStrapContext getBootstrapContext();
+
+ BufferAllocator getAllocator();
+
+ AuthenticatorProvider getAuthProvider();
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
index 00a324b..d337be0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
@@ -34,7 +34,7 @@ import org.apache.drill.exec.proto.UserBitShared.QueryResult;
import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
import org.apache.drill.exec.proto.helper.QueryIdHelper;
import org.apache.drill.exec.rpc.BaseRpcOutcomeListener;
-import org.apache.drill.exec.rpc.BasicClientWithConnection.ServerConnection;
+import org.apache.drill.exec.rpc.user.UserClient.UserToBitConnection;
import org.apache.drill.exec.rpc.ConnectionThrottle;
import org.apache.drill.exec.rpc.RpcBus;
import org.apache.drill.exec.rpc.RpcConnectionHandler;
@@ -73,8 +73,8 @@ public class QueryResultHandler {
return new SubmissionListener(resultsListener);
}
- public RpcConnectionHandler<ServerConnection> getWrappedConnectionHandler(
- final RpcConnectionHandler<ServerConnection> handler) {
+ public RpcConnectionHandler<UserToBitConnection> getWrappedConnectionHandler(
+ final RpcConnectionHandler<UserToBitConnection> handler) {
return new ChannelClosedHandler(handler);
}
@@ -350,20 +350,20 @@ public class QueryResultHandler {
}
/**
- * When a {@link ServerConnection connection} to a server is successfully created, this handler adds a
+ * When a {@link UserToBitConnection connection} to a server is successfully created, this handler adds a
* listener to that connection that listens to connection closure. If the connection is closed, all active
* {@link UserResultsListener result listeners} are failed.
*/
- private class ChannelClosedHandler implements RpcConnectionHandler<ServerConnection> {
+ private class ChannelClosedHandler implements RpcConnectionHandler<UserToBitConnection> {
- private final RpcConnectionHandler<ServerConnection> parentHandler;
+ private final RpcConnectionHandler<UserToBitConnection> parentHandler;
- public ChannelClosedHandler(final RpcConnectionHandler<ServerConnection> parentHandler) {
+ public ChannelClosedHandler(final RpcConnectionHandler<UserToBitConnection> parentHandler) {
this.parentHandler = parentHandler;
}
@Override
- public void connectionSucceeded(final ServerConnection connection) {
+ public void connectionSucceeded(final UserToBitConnection connection) {
connection.getChannel().closeFuture().addListener(
new GenericFutureListener<Future<? super Void>>() {
@Override
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
----------------------------------------------------------------------
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 c37e133..96b5669 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
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -17,16 +17,33 @@
*/
package org.apache.drill.exec.rpc.user;
+import com.google.common.base.Strings;
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.AbstractCheckedFuture;
+import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
+import io.netty.channel.socket.SocketChannel;
+import org.apache.drill.common.KerberosUtil;
+import org.apache.drill.common.config.DrillProperties;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
-import org.apache.drill.exec.proto.UserBitShared;
import org.apache.drill.exec.proto.UserBitShared.QueryData;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
import org.apache.drill.exec.proto.UserBitShared.QueryResult;
+import org.apache.drill.exec.proto.UserBitShared.SaslMessage;
+import org.apache.drill.exec.proto.UserBitShared.UserCredentials;
import org.apache.drill.exec.proto.UserProtos.BitToUserHandshake;
import org.apache.drill.exec.proto.UserProtos.CreatePreparedStatementResp;
import org.apache.drill.exec.proto.UserProtos.GetCatalogsResp;
@@ -34,49 +51,65 @@ import org.apache.drill.exec.proto.UserProtos.GetColumnsResp;
import org.apache.drill.exec.proto.UserProtos.GetQueryPlanFragments;
import org.apache.drill.exec.proto.UserProtos.GetSchemasResp;
import org.apache.drill.exec.proto.UserProtos.GetTablesResp;
-import org.apache.drill.exec.proto.UserProtos.HandshakeStatus;
import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
import org.apache.drill.exec.proto.UserProtos.RpcEndpointInfos;
import org.apache.drill.exec.proto.UserProtos.RpcType;
import org.apache.drill.exec.proto.UserProtos.RunQuery;
-import org.apache.drill.exec.proto.UserProtos.UserProperties;
+import org.apache.drill.exec.proto.UserProtos.SaslSupport;
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.BasicClientWithConnection;
-import org.apache.drill.exec.rpc.ConnectionThrottle;
+import org.apache.drill.exec.rpc.BasicClient;
+import org.apache.drill.exec.rpc.NonTransientRpcException;
+import org.apache.drill.exec.rpc.security.AuthenticationOutcomeListener;
import org.apache.drill.exec.rpc.DrillRpcFuture;
import org.apache.drill.exec.rpc.OutOfMemoryHandler;
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.RpcException;
-import org.apache.drill.exec.rpc.InvalidConnectionInfoException;
import com.google.protobuf.MessageLite;
+import org.apache.drill.exec.rpc.RpcOutcomeListener;
+import org.apache.drill.exec.rpc.security.AuthStringUtil;
+import org.apache.drill.exec.rpc.security.AuthenticatorFactory;
+import org.apache.drill.exec.rpc.security.plain.PlainFactory;
+import org.apache.drill.exec.rpc.security.ClientAuthenticatorProvider;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
-import io.netty.buffer.ByteBuf;
import io.netty.channel.EventLoopGroup;
+import org.slf4j.Logger;
-public class UserClient extends BasicClientWithConnection<RpcType, UserToBitHandshake, BitToUserHandshake> {
- private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserClient.class);
+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;
private final QueryResultHandler queryResultHandler = new QueryResultHandler();
private final String clientName;
- private RpcEndpointInfos serverInfos = null;
+ private RpcEndpointInfos serverInfos = null;
private boolean supportComplexTypes = true;
+ // 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 alloc, EventLoopGroup eventLoopGroup, Executor eventExecutor) {
+ BufferAllocator allocator, EventLoopGroup eventLoopGroup, Executor eventExecutor) {
super(
UserRpcConfig.getMapping(config, eventExecutor),
- alloc,
+ allocator.getAsByteBufAllocator(),
eventLoopGroup,
RpcType.HANDSHAKE,
BitToUserHandshake.class,
- BitToUserHandshake.PARSER,
- "user client");
+ BitToUserHandshake.PARSER);
this.clientName = clientName;
+ this.allocator = allocator;
this.supportComplexTypes = supportComplexTypes;
}
@@ -88,22 +121,169 @@ public class UserClient extends BasicClientWithConnection<RpcType, UserToBitHand
send(queryResultHandler.getWrappedListener(resultsListener), RpcType.RUN_QUERY, query, QueryId.class);
}
- public void connect(RpcConnectionHandler<ServerConnection> handler, DrillbitEndpoint endpoint,
- UserProperties props, UserBitShared.UserCredentials credentials) {
- UserToBitHandshake.Builder hsBuilder = UserToBitHandshake.newBuilder()
+ /**
+ * Connects, and if required, authenticates. This method blocks until both operations are complete.
+ *
+ * @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 handshake = UserToBitHandshake.newBuilder()
.setRpcVersion(UserRpcConfig.RPC_VERSION)
.setSupportListening(true)
.setSupportComplexTypes(supportComplexTypes)
.setSupportTimeout(true)
.setCredentials(credentials)
- .setClientInfos(UserRpcUtils.getRpcEndpointInfos(clientName));
+ .setClientInfos(UserRpcUtils.getRpcEndpointInfos(clientName))
+ .setSaslSupport(SaslSupport.SASL_AUTH)
+ .setProperties(properties.serializeForServer())
+ .build();
+
+ connect(handshake, endpoint).checkedGet();
+
+ if (serverAuthMechanisms != null) {
+ try {
+ authenticate(properties).checkedGet();
+ } catch (final SaslException e) {
+ throw new NonTransientRpcException(e);
+ }
+ }
+ }
+
+ private CheckedFuture<Void, RpcException> connect(final UserToBitHandshake handshake,
+ 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) {
+ return RpcException.mapException(e);
+ }
+ };
+ final RpcConnectionHandler<UserToBitConnection> connectionHandler =
+ new RpcConnectionHandler<UserToBitConnection>() {
+ @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));
+ }
+ };
+
+ connectAsClient(queryResultHandler.getWrappedConnectionHandler(connectionHandler),
+ handshake, endpoint.getAddress(), endpoint.getUserPort());
+
+ return connectionFuture;
+ }
+
+ private CheckedFuture<Void, SaslException> authenticate(final DrillProperties properties) {
+ final Map<String, String> propertiesMap = properties.stringPropertiesAsMap();
+
+ 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) {
+ if (e instanceof ExecutionException) {
+ final Throwable cause = Throwables.getRootCause(e);
+ if (cause instanceof SaslException) {
+ return new SaslException("Authentication failed: " + cause.getMessage(), cause);
+ }
+ }
+ return new SaslException("Authentication failed unexpectedly.", e);
+ }
+ };
+
+ final AuthenticatorFactory factory;
+ final String mechanismName;
+ final UserGroupInformation ugi;
+ final SaslClient saslClient;
+ try {
+ factory = getAuthenticatorFactory(properties);
+ mechanismName = factory.getSimpleName();
+ logger.trace("Will try to authenticate to server using {} mechanism.", mechanismName);
+ ugi = factory.createAndLoginUser(propertiesMap);
+ saslClient = factory.createSaslClient(ugi, propertiesMap);
+ if (saslClient == null) {
+ throw new SaslException("Cannot initiate authentication. Insufficient credentials?");
+ }
+ connection.setSaslClient(saslClient);
+ } catch (final IOException e) {
+ authSettable.setException(e);
+ return authFuture;
+ }
+
+ logger.trace("Initiating SASL exchange.");
+ new AuthenticationOutcomeListener<>(this, connection, RpcType.SASL_MESSAGE, ugi,
+ new RpcOutcomeListener<Void>() {
+ @Override
+ public void failed(RpcException ex) {
+ authSettable.setException(ex);
+ }
+
+ @Override
+ public void success(Void value, ByteBuf buffer) {
+ authComplete = true;
+ authSettable.set(null);
+ }
+
+ @Override
+ public void interrupted(InterruptedException e) {
+ authSettable.setException(e);
+ }
+ }).initiate(mechanismName);
+ return authFuture;
+ }
+
+ private AuthenticatorFactory getAuthenticatorFactory(final DrillProperties properties) throws SaslException {
+ final Set<String> mechanismSet = AuthStringUtil.asSet(serverAuthMechanisms);
- if (props != null) {
- hsBuilder.setProperties(props);
+ // first, check if a certain mechanism must be used
+ String authMechanism = properties.getProperty(DrillProperties.AUTH_MECHANISM);
+ if (authMechanism != null) {
+ if (!ClientAuthenticatorProvider.getInstance()
+ .containsFactory(authMechanism)) {
+ 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",
+ authMechanism));
+ }
+ return ClientAuthenticatorProvider.getInstance()
+ .getAuthenticatorFactory(authMechanism);
}
- this.connectAsClient(queryResultHandler.getWrappedConnectionHandler(handler),
- hsBuilder.build(), endpoint.getAddress(), endpoint.getUserPort());
+ // check if Kerberos is supported, and the service principal is provided
+ 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);
+ }
+
+ throw new SaslException(String.format("Server requires authentication using %s. Insufficient credentials?",
+ serverAuthMechanisms));
+ }
+
+ 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
@@ -131,19 +311,30 @@ public class UserClient extends BasicClientWithConnection<RpcType, UserToBitHand
return GetColumnsResp.getDefaultInstance();
case RpcType.PREPARED_STATEMENT_VALUE:
return CreatePreparedStatementResp.getDefaultInstance();
+ case RpcType.SASL_MESSAGE_VALUE:
+ return SaslMessage.getDefaultInstance();
}
throw new RpcException(String.format("Unable to deal with RpcType of %d", rpcType));
}
@Override
- protected Response handleReponse(ConnectionThrottle throttle, int rpcType, ByteBuf pBody, ByteBuf dBody) throws RpcException {
+ 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()));
+ }
switch (rpcType) {
case RpcType.QUERY_DATA_VALUE:
- queryResultHandler.batchArrived(throttle, pBody, dBody);
- return new Response(RpcType.ACK, Acks.OK);
+ queryResultHandler.batchArrived(connection, pBody, dBody);
+ sender.send(new Response(RpcType.ACK, Acks.OK));
+ break;
case RpcType.QUERY_RESULT_VALUE:
queryResultHandler.resultArrived(pBody);
- return new Response(RpcType.ACK, Acks.OK);
+ sender.send(new Response(RpcType.ACK, Acks.OK));
+ break;
default:
throw new RpcException(String.format("Unknown Rpc Type %d. ", rpcType));
}
@@ -155,16 +346,46 @@ public class UserClient extends BasicClientWithConnection<RpcType, UserToBitHand
if (inbound.hasServerInfos()) {
serverInfos = inbound.getServerInfos();
}
- if (inbound.getStatus() != HandshakeStatus.SUCCESS) {
+ switch (inbound.getStatus()) {
+ case SUCCESS:
+ break;
+ case AUTH_REQUIRED: {
+ authComplete = false;
+ logger.trace("Server requires authentication before proceeding.");
+ serverAuthMechanisms = ImmutableList.copyOf(inbound.getAuthenticationMechanismsList());
+ 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 InvalidConnectionInfoException(errMsg);
+ throw new NonTransientRpcException(errMsg);
}
}
@Override
- protected void finalizeConnection(BitToUserHandshake handshake, BasicClientWithConnection.ServerConnection connection) {
+ protected UserToBitConnection initRemoteConnection(SocketChannel channel) {
+ super.initRemoteConnection(channel);
+ return new UserToBitConnection(channel);
+ }
+
+ public class UserToBitConnection extends AbstractClientConnection {
+
+ UserToBitConnection(SocketChannel channel) {
+ super(channel, "user client");
+ }
+
+ @Override
+ public BufferAllocator getAllocator() {
+ return allocator;
+ }
+
+ @Override
+ protected Logger getLogger() {
+ return logger;
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserConnectionConfig.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..49a866b
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserConnectionConfig.java
@@ -0,0 +1,74 @@
+/*
+ * 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.rpc.user;
+
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.exception.DrillbitStartupException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.rpc.AbstractConnectionConfig;
+import org.apache.drill.exec.rpc.RequestHandler;
+import org.apache.drill.exec.server.BootStrapContext;
+
+// config for bit to user connection
+// package private
+class UserConnectionConfig extends AbstractConnectionConfig {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserConnectionConfig.class);
+
+ private final boolean authEnabled;
+ private final InboundImpersonationManager impersonationManager;
+
+ private final UserServerRequestHandler handler;
+
+ UserConnectionConfig(BufferAllocator allocator, BootStrapContext context, UserServerRequestHandler handler)
+ throws DrillbitStartupException {
+ super(allocator, context);
+ this.handler = handler;
+
+ if (context.getConfig().getBoolean(ExecConstants.USER_AUTHENTICATION_ENABLED)) {
+ if (getAuthProvider().getAllFactoryNames().isEmpty()) {
+ throw new DrillbitStartupException("Authentication enabled, but no mechanisms found. Please check " +
+ "authentication configuration.");
+ }
+ authEnabled = true;
+ logger.info("Configured all user connections to require authentication using: {}",
+ getAuthProvider().getAllFactoryNames());
+ } else {
+ authEnabled = false;
+ }
+
+ impersonationManager = !context.getConfig().getBoolean(ExecConstants.IMPERSONATION_ENABLED) ? null :
+ new InboundImpersonationManager();
+ }
+
+ @Override
+ public String getName() {
+ return "user server";
+ }
+
+ boolean isAuthEnabled() {
+ return authEnabled;
+ }
+
+ InboundImpersonationManager getImpersonationManager() {
+ return impersonationManager;
+ }
+
+ RequestHandler<UserServer.BitToUserConnection> getMessageHandler() {
+ return handler;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserRpcConfig.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserRpcConfig.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserRpcConfig.java
index c520744..645ded5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserRpcConfig.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserRpcConfig.java
@@ -25,6 +25,7 @@ import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
import org.apache.drill.exec.proto.UserBitShared.QueryData;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
import org.apache.drill.exec.proto.UserBitShared.QueryResult;
+import org.apache.drill.exec.proto.UserBitShared.SaslMessage;
import org.apache.drill.exec.proto.UserProtos.BitToUserHandshake;
import org.apache.drill.exec.proto.UserProtos.CreatePreparedStatementReq;
import org.apache.drill.exec.proto.UserProtos.CreatePreparedStatementResp;
@@ -44,7 +45,7 @@ import org.apache.drill.exec.proto.UserProtos.UserToBitHandshake;
import org.apache.drill.exec.rpc.RpcConfig;
public class UserRpcConfig {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserRpcConfig.class);
+// private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserRpcConfig.class);
public static RpcConfig getMapping(DrillConfig config, Executor executor) {
return RpcConfig.newBuilder()
@@ -65,8 +66,13 @@ public class UserRpcConfig {
.add(RpcType.GET_COLUMNS, GetColumnsReq.class, RpcType.COLUMNS, GetColumnsResp.class) // user to bit
.add(RpcType.CREATE_PREPARED_STATEMENT, CreatePreparedStatementReq.class,
RpcType.PREPARED_STATEMENT, CreatePreparedStatementResp.class) // user to bit
+ .add(RpcType.SASL_MESSAGE, SaslMessage.class, RpcType.SASL_MESSAGE, SaslMessage.class) // user <-> bit
.build();
}
- public static int RPC_VERSION = 5;
+ public static final int RPC_VERSION = 5;
+
+ // prevent instantiation
+ private UserRpcConfig() {
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
----------------------------------------------------------------------
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 281b124..6854a3e 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
@@ -20,83 +20,64 @@ package org.apache.drill.exec.rpc.user;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.UUID;
-import java.util.concurrent.Executor;
-import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.common.scanner.persistence.ScanResult;
-import org.apache.drill.exec.ExecConstants;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.drill.common.config.DrillProperties;
import org.apache.drill.exec.exception.DrillbitStartupException;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.physical.impl.materialize.QueryWritableBatch;
import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
import org.apache.drill.exec.proto.GeneralRPCProtos.RpcMode;
-import org.apache.drill.exec.proto.UserBitShared.QueryId;
import org.apache.drill.exec.proto.UserBitShared.QueryResult;
+import org.apache.drill.exec.proto.UserBitShared.UserCredentials;
import org.apache.drill.exec.proto.UserProtos.BitToUserHandshake;
-import org.apache.drill.exec.proto.UserProtos.CreatePreparedStatementReq;
-import org.apache.drill.exec.proto.UserProtos.GetCatalogsReq;
-import org.apache.drill.exec.proto.UserProtos.GetColumnsReq;
-import org.apache.drill.exec.proto.UserProtos.GetQueryPlanFragments;
-import org.apache.drill.exec.proto.UserProtos.GetSchemasReq;
-import org.apache.drill.exec.proto.UserProtos.GetTablesReq;
import org.apache.drill.exec.proto.UserProtos.HandshakeStatus;
import org.apache.drill.exec.proto.UserProtos.Property;
import org.apache.drill.exec.proto.UserProtos.RpcType;
-import org.apache.drill.exec.proto.UserProtos.RunQuery;
+import org.apache.drill.exec.proto.UserProtos.SaslSupport;
import org.apache.drill.exec.proto.UserProtos.UserProperties;
import org.apache.drill.exec.proto.UserProtos.UserToBitHandshake;
+import org.apache.drill.exec.rpc.AbstractRemoteConnection;
import org.apache.drill.exec.rpc.BasicServer;
import org.apache.drill.exec.rpc.OutOfMemoryHandler;
import org.apache.drill.exec.rpc.OutboundRpcMessage;
import org.apache.drill.exec.rpc.ProtobufLengthDecoder;
-import org.apache.drill.exec.rpc.RemoteConnection;
-import org.apache.drill.exec.rpc.Response;
-import org.apache.drill.exec.rpc.ResponseSender;
import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.rpc.RpcOutcomeListener;
-import org.apache.drill.exec.rpc.user.UserServer.UserClientConnectionImpl;
+import org.apache.drill.exec.rpc.AbstractServerConnection;
+import org.apache.drill.exec.rpc.security.ServerAuthenticationHandler;
+import org.apache.drill.exec.rpc.user.UserServer.BitToUserConnection;
+import org.apache.drill.exec.rpc.security.plain.PlainFactory;
import org.apache.drill.exec.rpc.user.security.UserAuthenticationException;
-import org.apache.drill.exec.rpc.user.security.UserAuthenticator;
-import org.apache.drill.exec.rpc.user.security.UserAuthenticatorFactory;
+import org.apache.drill.exec.server.BootStrapContext;
import org.apache.drill.exec.work.user.UserWorker;
-import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.MessageLite;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufInputStream;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
+import org.apache.hadoop.security.HadoopKerberosName;
+import org.slf4j.Logger;
-public class UserServer extends BasicServer<RpcType, UserClientConnectionImpl> {
+import javax.security.sasl.SaslException;
+
+public class UserServer extends BasicServer<RpcType, BitToUserConnection> {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserServer.class);
private static final String SERVER_NAME = "Apache Drill Server";
- final UserWorker worker;
- final BufferAllocator alloc;
- final UserAuthenticator authenticator;
- final InboundImpersonationManager impersonationManager;
+ private final UserConnectionConfig config;
+ private final UserWorker userWorker;
- public UserServer(DrillConfig config, ScanResult classpathScan, BufferAllocator alloc, EventLoopGroup eventLoopGroup,
- UserWorker worker, Executor executor) throws DrillbitStartupException {
- super(UserRpcConfig.getMapping(config, executor),
- alloc.getAsByteBufAllocator(),
+ public UserServer(BootStrapContext context, BufferAllocator allocator, EventLoopGroup eventLoopGroup,
+ UserWorker worker) throws DrillbitStartupException {
+ super(UserRpcConfig.getMapping(context.getConfig(), context.getExecutor()),
+ allocator.getAsByteBufAllocator(),
eventLoopGroup);
- this.worker = worker;
- this.alloc = alloc;
- // TODO: move this up
- if (config.getBoolean(ExecConstants.USER_AUTHENTICATION_ENABLED)) {
- authenticator = UserAuthenticatorFactory.createAuthenticator(config, classpathScan);
- } else {
- authenticator = null;
- }
- if (config.getBoolean(ExecConstants.IMPERSONATION_ENABLED)) {
- impersonationManager = new InboundImpersonationManager();
- } else {
- impersonationManager = null;
- }
+ this.config = new UserConnectionConfig(allocator, context, new UserServerRequestHandler(worker));
+ this.userWorker = worker;
}
@Override
@@ -110,98 +91,9 @@ public class UserServer extends BasicServer<RpcType, UserClientConnectionImpl> {
}
}
- @Override
- protected void handle(UserClientConnectionImpl connection, int rpcType, ByteBuf pBody, ByteBuf dBody,
- ResponseSender responseSender) throws RpcException {
- switch (rpcType) {
-
- case RpcType.RUN_QUERY_VALUE:
- logger.debug("Received query to run. Returning query handle.");
- try {
- final RunQuery query = RunQuery.PARSER.parseFrom(new ByteBufInputStream(pBody));
- final QueryId queryId = worker.submitWork(connection, query);
- responseSender.send(new Response(RpcType.QUERY_HANDLE, queryId));
- break;
- } catch (InvalidProtocolBufferException e) {
- throw new RpcException("Failure while decoding RunQuery body.", e);
- }
-
- case RpcType.CANCEL_QUERY_VALUE:
- try {
- final QueryId queryId = QueryId.PARSER.parseFrom(new ByteBufInputStream(pBody));
- final Ack ack = worker.cancelQuery(queryId);
- responseSender.send(new Response(RpcType.ACK, ack));
- break;
- } catch (InvalidProtocolBufferException e) {
- throw new RpcException("Failure while decoding QueryId body.", e);
- }
-
- case RpcType.RESUME_PAUSED_QUERY_VALUE:
- try {
- final QueryId queryId = QueryId.PARSER.parseFrom(new ByteBufInputStream(pBody));
- final Ack ack = worker.resumeQuery(queryId);
- responseSender.send(new Response(RpcType.ACK, ack));
- break;
- } catch (final InvalidProtocolBufferException e) {
- throw new RpcException("Failure while decoding QueryId body.", e);
- }
- case RpcType.GET_QUERY_PLAN_FRAGMENTS_VALUE:
- try {
- final GetQueryPlanFragments req = GetQueryPlanFragments.PARSER.parseFrom(new ByteBufInputStream(pBody));
- responseSender.send(new Response(RpcType.QUERY_PLAN_FRAGMENTS, worker.getQueryPlan(connection, req)));
- break;
- } catch(final InvalidProtocolBufferException e) {
- throw new RpcException("Failure while decoding GetQueryPlanFragments body.", e);
- }
- case RpcType.GET_CATALOGS_VALUE:
- try {
- final GetCatalogsReq req = GetCatalogsReq.PARSER.parseFrom(new ByteBufInputStream(pBody));
- worker.submitCatalogMetadataWork(connection.getSession(), req, responseSender);
- break;
- } catch (final InvalidProtocolBufferException e) {
- throw new RpcException("Failure while decoding GetCatalogsReq body.", e);
- }
- case RpcType.GET_SCHEMAS_VALUE:
- try {
- final GetSchemasReq req = GetSchemasReq.PARSER.parseFrom(new ByteBufInputStream(pBody));
- worker.submitSchemasMetadataWork(connection.getSession(), req, responseSender);
- break;
- } catch (final InvalidProtocolBufferException e) {
- throw new RpcException("Failure while decoding GetSchemasReq body.", e);
- }
- case RpcType.GET_TABLES_VALUE:
- try {
- final GetTablesReq req = GetTablesReq.PARSER.parseFrom(new ByteBufInputStream(pBody));
- worker.submitTablesMetadataWork(connection.getSession(), req, responseSender);
- break;
- } catch (final InvalidProtocolBufferException e) {
- throw new RpcException("Failure while decoding GetTablesReq body.", e);
- }
- case RpcType.GET_COLUMNS_VALUE:
- try {
- final GetColumnsReq req = GetColumnsReq.PARSER.parseFrom(new ByteBufInputStream(pBody));
- worker.submitColumnsMetadataWork(connection.getSession(), req, responseSender);
- break;
- } catch (final InvalidProtocolBufferException e) {
- throw new RpcException("Failure while decoding GetColumnsReq body.", e);
- }
- case RpcType.CREATE_PREPARED_STATEMENT_VALUE:
- try {
- final CreatePreparedStatementReq req =
- CreatePreparedStatementReq.PARSER.parseFrom(new ByteBufInputStream(pBody));
- worker.submitPreparedStatementWork(connection, req, responseSender);
- break;
- } catch (final InvalidProtocolBufferException e) {
- throw new RpcException("Failure while decoding CreatePreparedStatementReq body.", e);
- }
- default:
- throw new UnsupportedOperationException(String.format("UserServer received rpc of unknown type. Type was %d.", rpcType));
- }
- }
-
/**
* Interface for getting user session properties and interacting with user connection. Separating this interface from
- * {@link RemoteConnection} implementation for user connection:
+ * {@link AbstractRemoteConnection} implementation for user connection:
* <p><ul>
* <li> Connection is passed to Foreman and Screen operators. Instead passing this interface exposes few details.
* <li> Makes it easy to have wrappers around user connection which can be helpful to tap the messages and data
@@ -241,30 +133,58 @@ public class UserServer extends BasicServer<RpcType, UserClientConnectionImpl> {
}
/**
- * {@link RemoteConnection} implementation for user connection. Also implements {@link UserClientConnection}.
+ * {@link AbstractRemoteConnection} implementation for user connection. Also implements {@link UserClientConnection}.
*/
- public class UserClientConnectionImpl extends RemoteConnection implements UserClientConnection {
+ public class BitToUserConnection extends AbstractServerConnection<BitToUserConnection>
+ implements UserClientConnection {
private UserSession session;
+ private UserToBitHandshake inbound;
- public UserClientConnectionImpl(SocketChannel channel) {
- super(channel, "user client");
+ BitToUserConnection(SocketChannel channel) {
+ super(channel, config, !config.isAuthEnabled()
+ ? config.getMessageHandler()
+ : new ServerAuthenticationHandler<>(config.getMessageHandler(),
+ RpcType.SASL_MESSAGE_VALUE, RpcType.SASL_MESSAGE));
}
void disableReadTimeout() {
getChannel().pipeline().remove(BasicServer.TIMEOUT_HANDLER);
}
- void setUser(final UserToBitHandshake inbound) throws IOException {
+ void setHandshake(final UserToBitHandshake inbound) {
+ this.inbound = inbound;
+ }
+
+ @Override
+ public void finalizeSaslSession() throws IOException {
+ final String authorizationID = getSaslServer().getAuthorizationID();
+ final String userName = new HadoopKerberosName(authorizationID).getShortName();
+ logger.debug("Created session for {}", userName);
+ finalizeSession(userName);
+ }
+
+ /**
+ * Sets the user on the session, and finalizes the session.
+ *
+ * @param userName user name to set on the session
+ *
+ */
+ void finalizeSession(String userName) {
+ // create a session
session = UserSession.Builder.newBuilder()
- .withCredentials(inbound.getCredentials())
- .withOptionManager(worker.getSystemOptions())
+ .withCredentials(UserCredentials.newBuilder()
+ .setUserName(userName)
+ .build())
+ .withOptionManager(userWorker.getSystemOptions())
.withUserProperties(inbound.getProperties())
.setSupportComplexTypes(inbound.getSupportComplexTypes())
.build();
+
+ // if inbound impersonation is enabled and a target is mentioned
final String targetName = session.getTargetUserName();
- if (impersonationManager != null && targetName != null) {
- impersonationManager.replaceUserOnSession(targetName, session);
+ if (config.getImpersonationManager() != null && targetName != null) {
+ config.getImpersonationManager().replaceUserOnSession(targetName, session);
}
}
@@ -286,13 +206,19 @@ public class UserServer extends BasicServer<RpcType, UserClientConnectionImpl> {
}
@Override
- public BufferAllocator getAllocator() {
- return alloc;
+ protected Logger getLogger() {
+ return logger;
}
@Override
public ChannelFuture getChannelClosureFuture() {
- return getChannel().closeFuture();
+ return getChannel().closeFuture()
+ .addListener(new GenericFutureListener<Future<? super Void>>() {
+ @Override
+ public void operationComplete(Future<? super Void> future) throws Exception {
+ cleanup();
+ }
+ });
}
@Override
@@ -300,26 +226,27 @@ public class UserServer extends BasicServer<RpcType, UserClientConnectionImpl> {
return getChannel().remoteAddress();
}
- @Override
- public void closeSession() {
- session.close();
+ private void cleanup() {
+ if (session != null) {
+ session.close();
+ }
}
@Override
public void close() {
- closeSession();
+ cleanup();
super.close();
}
}
@Override
- public UserClientConnectionImpl initRemoteConnection(SocketChannel channel) {
+ protected BitToUserConnection initRemoteConnection(SocketChannel channel) {
super.initRemoteConnection(channel);
- return new UserClientConnectionImpl(channel);
+ return new BitToUserConnection(channel);
}
@Override
- protected ServerHandshakeHandler<UserToBitHandshake> getHandshakeHandler(final UserClientConnectionImpl connection) {
+ protected ServerHandshakeHandler<UserToBitHandshake> getHandshakeHandler(final BitToUserConnection connection) {
return new ServerHandshakeHandler<UserToBitHandshake>(RpcType.HANDSHAKE, UserToBitHandshake.PARSER){
@@ -329,7 +256,8 @@ public class UserServer extends BasicServer<RpcType, UserClientConnectionImpl> {
OutboundRpcMessage msg = new OutboundRpcMessage(RpcMode.RESPONSE, this.handshakeType, coordinationId, handshakeResp);
ctx.writeAndFlush(msg);
- if (handshakeResp.getStatus() != HandshakeStatus.SUCCESS) {
+ if (handshakeResp.getStatus() != HandshakeStatus.SUCCESS &&
+ handshakeResp.getStatus() != HandshakeStatus.AUTH_REQUIRED) {
// If handling handshake results in an error, throw an exception to terminate the connection.
throw new RpcException("Handshake request failed: " + handshakeResp.getErrorMessage());
}
@@ -339,7 +267,6 @@ public class UserServer extends BasicServer<RpcType, UserClientConnectionImpl> {
public BitToUserHandshake getHandshakeResponse(UserToBitHandshake inbound) throws Exception {
logger.trace("Handling handshake from user to bit. {}", inbound);
-
// if timeout is unsupported or is set to false, disable timeout.
if (!inbound.hasSupportTimeout() || !inbound.getSupportTimeout()) {
connection.disableReadTimeout();
@@ -358,25 +285,62 @@ public class UserServer extends BasicServer<RpcType, UserClientConnectionImpl> {
return handleFailure(respBuilder, HandshakeStatus.RPC_VERSION_MISMATCH, errMsg, null);
}
- if (authenticator != null) {
+ connection.setHandshake(inbound);
+
+ if (!config.isAuthEnabled()) {
+ connection.finalizeSession(inbound.getCredentials().getUserName());
+ respBuilder.setStatus(HandshakeStatus.SUCCESS);
+ return respBuilder.build();
+ }
+
+ final boolean clientSupportsSasl = inbound.hasSaslSupport() &&
+ (inbound.getSaslSupport().ordinal() >= SaslSupport.SASL_AUTH.ordinal());
+ if (!clientSupportsSasl) { // for backward compatibility < 1.10
+ final String userName = inbound.getCredentials().getUserName();
+ if (logger.isTraceEnabled()) {
+ logger.trace("User {} on connection {} is likely using an older client.",
+ userName, connection.getRemoteAddress());
+ }
try {
String password = "";
final UserProperties props = inbound.getProperties();
for (int i = 0; i < props.getPropertiesCount(); i++) {
Property prop = props.getProperties(i);
- if (UserSession.PASSWORD.equalsIgnoreCase(prop.getKey())) {
+ if (DrillProperties.PASSWORD.equalsIgnoreCase(prop.getKey())) {
password = prop.getValue();
break;
}
}
- authenticator.authenticate(inbound.getCredentials().getUserName(), password);
+ final PlainFactory plainFactory;
+ try {
+ plainFactory = (PlainFactory) config.getAuthProvider()
+ .getAuthenticatorFactory(PlainFactory.SIMPLE_NAME);
+ } catch (final SaslException e) {
+ throw new UserAuthenticationException("The server no longer supports username/password" +
+ " based authentication. Please talk to your system administrator.");
+ }
+ plainFactory.getAuthenticator()
+ .authenticate(userName, password);
+ connection.changeHandlerTo(config.getMessageHandler());
+ connection.finalizeSession(userName);
+ respBuilder.setStatus(HandshakeStatus.SUCCESS);
+ if (logger.isTraceEnabled()) {
+ logger.trace("Authenticated {} successfully using PLAIN from {}", userName,
+ connection.getRemoteAddress());
+ }
+ return respBuilder.build();
} catch (UserAuthenticationException ex) {
return handleFailure(respBuilder, HandshakeStatus.AUTH_FAILED, ex.getMessage(), ex);
}
}
- connection.setUser(inbound);
- return respBuilder.setStatus(HandshakeStatus.SUCCESS).build();
+ // mention server's authentication capabilities
+ respBuilder.addAllAuthenticationMechanisms(config.getAuthProvider().getAllFactoryNames());
+
+ // for now, this means PLAIN credentials will be sent over twice
+ // (during handshake and during sasl exchange)
+ respBuilder.setStatus(HandshakeStatus.AUTH_REQUIRED);
+ return respBuilder.build();
} catch (Exception e) {
return handleFailure(respBuilder, HandshakeStatus.UNKNOWN_FAILURE, e.getMessage(), e);
}
@@ -412,19 +376,8 @@ public class UserServer extends BasicServer<RpcType, UserClientConnectionImpl> {
}
@Override
- public ProtobufLengthDecoder getDecoder(BufferAllocator allocator, OutOfMemoryHandler outOfMemoryHandler) {
+ protected ProtobufLengthDecoder getDecoder(BufferAllocator allocator, OutOfMemoryHandler outOfMemoryHandler) {
return new UserProtobufLengthDecoder(allocator, outOfMemoryHandler);
}
- @Override
- public void close() throws IOException {
- try {
- if (authenticator != null) {
- authenticator.close();
- }
- } catch (Exception e) {
- logger.warn("Failure closing authenticator.", e);
- }
- super.close();
- }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/b6e59ec3/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServerRequestHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServerRequestHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServerRequestHandler.java
new file mode 100644
index 0000000..1c44176
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServerRequestHandler.java
@@ -0,0 +1,143 @@
+/*
+ * 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.rpc.user;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufInputStream;
+import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserProtos.CreatePreparedStatementReq;
+import org.apache.drill.exec.proto.UserProtos.GetCatalogsReq;
+import org.apache.drill.exec.proto.UserProtos.GetColumnsReq;
+import org.apache.drill.exec.proto.UserProtos.GetQueryPlanFragments;
+import org.apache.drill.exec.proto.UserProtos.GetSchemasReq;
+import org.apache.drill.exec.proto.UserProtos.GetTablesReq;
+import org.apache.drill.exec.proto.UserProtos.RpcType;
+import org.apache.drill.exec.proto.UserProtos.RunQuery;
+import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.ResponseSender;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.RequestHandler;
+import org.apache.drill.exec.rpc.user.UserServer.BitToUserConnection;
+import org.apache.drill.exec.work.user.UserWorker;
+
+/**
+ * Should create only one instance of this class per Drillbit service.
+ */
+// package private
+class UserServerRequestHandler implements RequestHandler<BitToUserConnection> {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserServerRequestHandler.class);
+
+ private final UserWorker worker;
+
+ public UserServerRequestHandler(final UserWorker worker) {
+ this.worker = worker;
+ }
+
+ @Override
+ public void handle(BitToUserConnection connection, int rpcType, ByteBuf pBody, ByteBuf dBody,
+ ResponseSender responseSender)
+ throws RpcException {
+ switch (rpcType) {
+
+ case RpcType.RUN_QUERY_VALUE:
+ logger.debug("Received query to run. Returning query handle.");
+ try {
+ final RunQuery query = RunQuery.PARSER.parseFrom(new ByteBufInputStream(pBody));
+ final QueryId queryId = worker.submitWork(connection, query);
+ responseSender.send(new Response(RpcType.QUERY_HANDLE, queryId));
+ break;
+ } catch (InvalidProtocolBufferException e) {
+ throw new RpcException("Failure while decoding RunQuery body.", e);
+ }
+
+ case RpcType.CANCEL_QUERY_VALUE:
+ try {
+ final QueryId queryId = QueryId.PARSER.parseFrom(new ByteBufInputStream(pBody));
+ final Ack ack = worker.cancelQuery(queryId);
+ responseSender.send(new Response(RpcType.ACK, ack));
+ break;
+ } catch (InvalidProtocolBufferException e) {
+ throw new RpcException("Failure while decoding QueryId body.", e);
+ }
+
+ case RpcType.RESUME_PAUSED_QUERY_VALUE:
+ try {
+ final QueryId queryId = QueryId.PARSER.parseFrom(new ByteBufInputStream(pBody));
+ final Ack ack = worker.resumeQuery(queryId);
+ responseSender.send(new Response(RpcType.ACK, ack));
+ break;
+ } catch (final InvalidProtocolBufferException e) {
+ throw new RpcException("Failure while decoding QueryId body.", e);
+ }
+ case RpcType.GET_QUERY_PLAN_FRAGMENTS_VALUE:
+ try {
+ final GetQueryPlanFragments req = GetQueryPlanFragments.PARSER.parseFrom(new ByteBufInputStream(pBody));
+ responseSender.send(new Response(RpcType.QUERY_PLAN_FRAGMENTS, worker.getQueryPlan(connection, req)));
+ break;
+ } catch(final InvalidProtocolBufferException e) {
+ throw new RpcException("Failure while decoding GetQueryPlanFragments body.", e);
+ }
+ case RpcType.GET_CATALOGS_VALUE:
+ try {
+ final GetCatalogsReq req = GetCatalogsReq.PARSER.parseFrom(new ByteBufInputStream(pBody));
+ worker.submitCatalogMetadataWork(connection.getSession(), req, responseSender);
+ break;
+ } catch (final InvalidProtocolBufferException e) {
+ throw new RpcException("Failure while decoding GetCatalogsReq body.", e);
+ }
+ case RpcType.GET_SCHEMAS_VALUE:
+ try {
+ final GetSchemasReq req = GetSchemasReq.PARSER.parseFrom(new ByteBufInputStream(pBody));
+ worker.submitSchemasMetadataWork(connection.getSession(), req, responseSender);
+ break;
+ } catch (final InvalidProtocolBufferException e) {
+ throw new RpcException("Failure while decoding GetSchemasReq body.", e);
+ }
+ case RpcType.GET_TABLES_VALUE:
+ try {
+ final GetTablesReq req = GetTablesReq.PARSER.parseFrom(new ByteBufInputStream(pBody));
+ worker.submitTablesMetadataWork(connection.getSession(), req, responseSender);
+ break;
+ } catch (final InvalidProtocolBufferException e) {
+ throw new RpcException("Failure while decoding GetTablesReq body.", e);
+ }
+ case RpcType.GET_COLUMNS_VALUE:
+ try {
+ final GetColumnsReq req = GetColumnsReq.PARSER.parseFrom(new ByteBufInputStream(pBody));
+ worker.submitColumnsMetadataWork(connection.getSession(), req, responseSender);
+ break;
+ } catch (final InvalidProtocolBufferException e) {
+ throw new RpcException("Failure while decoding GetColumnsReq body.", e);
+ }
+ case RpcType.CREATE_PREPARED_STATEMENT_VALUE:
+ try {
+ final CreatePreparedStatementReq req =
+ CreatePreparedStatementReq.PARSER.parseFrom(new ByteBufInputStream(pBody));
+ worker.submitPreparedStatementWork(connection, req, responseSender);
+ break;
+ } catch (final InvalidProtocolBufferException e) {
+ throw new RpcException("Failure while decoding CreatePreparedStatementReq body.", e);
+ }
+ default:
+ throw new UnsupportedOperationException(
+ String.format("UserServerRequestHandler received rpc of unknown type. Type was %d.", rpcType));
+ }
+ }
+}