[
https://issues.apache.org/jira/browse/DRILL-4280?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15625919#comment-15625919
]
ASF GitHub Bot commented on DRILL-4280:
---------------------------------------
Github user sudheeshkatkam commented on a diff in the pull request:
https://github.com/apache/drill/pull/578#discussion_r85853794
--- Diff:
exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java ---
@@ -78,21 +101,241 @@ public void submitQuery(UserResultsListener
resultsListener, RunQuery query) {
send(queryResultHandler.getWrappedListener(resultsListener),
RpcType.RUN_QUERY, query, QueryId.class);
}
- public void connect(RpcConnectionHandler<ServerConnection> handler,
DrillbitEndpoint endpoint,
- UserProperties props, UserBitShared.UserCredentials
credentials) {
+ public CheckedFuture<Void, RpcException> connect(DrillbitEndpoint
endpoint, ConnectionParameters parameters,
+ UserCredentials
credentials) {
+ final FutureHandler handler = new FutureHandler();
UserToBitHandshake.Builder hsBuilder = UserToBitHandshake.newBuilder()
.setRpcVersion(UserRpcConfig.RPC_VERSION)
.setSupportListening(true)
.setSupportComplexTypes(supportComplexTypes)
.setSupportTimeout(true)
- .setCredentials(credentials);
+ .setCredentials(credentials)
+ .setProperties(parameters.serializeForServer());
+ this.parameters = parameters;
+
+
connectAsClient(queryResultHandler.getWrappedConnectionHandler(handler),
+ hsBuilder.build(), endpoint.getAddress(), endpoint.getUserPort());
+ return handler;
+ }
+
+ /**
+ * Check (after {@link #connect connecting}) if server requires
authentication.
+ *
+ * @return true if server requires authentication
+ */
+ public boolean serverRequiresAuthentication() {
+ return supportedAuthMechs != null;
+ }
+
+ /**
+ * Returns a list of supported authentication mechanism. If called
before {@link #connect connecting},
+ * returns null. If called after {@link #connect connecting}, returns a
list of supported mechanisms
+ * iff authentication is required.
+ *
+ * @return list of supported authentication mechanisms
+ */
+ public List<String> getSupportedAuthenticationMechanisms() {
+ return supportedAuthMechs;
+ }
+
+ /**
+ * Authenticate to the server asynchronously. Returns a future that
{@link CheckedFuture#checkedGet results}
+ * in null if authentication succeeds, or throws a {@link SaslException}
with relevant message if
+ * authentication fails.
+ *
+ * This method uses parameters provided at {@link #connect connection
time} and override them with the
+ * given parameters, if any.
+ *
+ * @param overrides parameter overrides
+ * @return result of authentication request
+ */
+ public CheckedFuture<Void, SaslException> authenticate(final
ConnectionParameters overrides) {
+ if (supportedAuthMechs == null) {
+ throw new IllegalStateException("Server does not require
authentication.");
+ }
+ parameters.merge(overrides);
+
+ final SettableFuture<Void> settableFuture = SettableFuture.create();
// future used in SASL exchange
+ final CheckedFuture<Void, SaslException> future =
+ new AbstractCheckedFuture<Void, SaslException>(settableFuture) {
+
+ @Override
+ protected SaslException mapException(Exception e) {
+ if (connection != null) {
+ connection.close(); // to ensure connection is dropped
+ }
+ if (e instanceof ExecutionException) {
+ final Throwable cause = e.getCause();
+ if (cause instanceof SaslException) {
+ return new SaslException("Authentication failed: " +
cause.getMessage(), cause);
+ }
+ }
+ return new SaslException("Authentication failed
unexpectedly.", e);
+ }
+ };
- if (props != null) {
- hsBuilder.setProperties(props);
+ final ClientAuthenticationProvider authenticationProvider;
+ try {
+ authenticationProvider =
+
UserAuthenticationUtil.getClientAuthenticationProvider(parameters,
supportedAuthMechs);
+ } catch (final SaslException e) {
+ settableFuture.setException(e);
+ return future;
}
-
this.connectAsClient(queryResultHandler.getWrappedConnectionHandler(handler),
- hsBuilder.build(), endpoint.getAddress(), endpoint.getUserPort());
+ final String providerName = authenticationProvider.name();
+ logger.trace("Will try to login for {} mechanism.", providerName);
+ final UserGroupInformation ugi;
+ try {
+ ugi = authenticationProvider.login(parameters);
+ } catch (final SaslException e) {
+ settableFuture.setException(e);
+ return future;
+ }
+
+ logger.trace("Will try to authenticate to server using {} mechanism.",
providerName);
+ try {
+ saslClient = authenticationProvider.createSaslClient(ugi,
parameters);
+ } catch (final SaslException e) {
+ settableFuture.setException(e);
+ return future;
+ }
+
+ if (saslClient == null) {
+ settableFuture.setException(new SaslException("Cannot initiate
authentication. Insufficient credentials?"));
+ return future;
+ }
+ logger.trace("Initiating SASL exchange.");
+
+ try {
+ final ByteString responseData;
+ if (saslClient.hasInitialResponse()) {
+ responseData = ByteString.copyFrom(evaluateChallenge(ugi,
saslClient, new byte[0]));
+ } else {
+ responseData = ByteString.EMPTY;
+ }
+ send(new SaslChallengeHandler(ugi, settableFuture),
+ RpcType.SASL_MESSAGE,
+ SaslMessage.newBuilder()
+ .setMechanism(providerName)
+ .setStatus(SaslStatus.SASL_START)
+ .setData(responseData)
+ .build(),
+ SaslMessage.class);
+ logger.trace("Initiated SASL exchange.");
+ } catch (final SaslException e) {
+ settableFuture.setException(e);
+ }
+ return future;
+ }
+
+ private static byte[] evaluateChallenge(final UserGroupInformation ugi,
final SaslClient saslClient,
+ final byte[] challenge) throws
SaslException {
+ try {
+ return ugi.doAs(new PrivilegedExceptionAction<byte[]>() {
+ @Override
+ public byte[] run() throws Exception {
+ return saslClient.evaluateChallenge(challenge);
+ }
+ });
+ } catch (final UndeclaredThrowableException e) {
+ final Throwable cause = e.getCause();
+ if (cause instanceof SaslException) {
+ throw (SaslException) cause;
+ } else {
+ throw new SaslException(
+ String.format("Unexpected failure (%s)",
saslClient.getMechanismName()), cause);
+ }
+ } catch (final IOException | InterruptedException e) {
+ throw new SaslException(String.format("Unexpected failure (%s)",
saslClient.getMechanismName()), e);
+ }
+ }
+
+ // handles SASL message exchange
+ private class SaslChallengeHandler implements
RpcOutcomeListener<SaslMessage> {
+
+ private final UserGroupInformation ugi;
+ private final SettableFuture<Void> future;
+
+ public SaslChallengeHandler(UserGroupInformation ugi,
SettableFuture<Void> future) {
+ this.ugi = ugi;
+ this.future = future;
+ }
+
+ @Override
+ public void failed(RpcException ex) {
+ future.setException(new SaslException("Unexpected failure", ex));
+ }
+
+ @Override
+ public void success(SaslMessage value, ByteBuf buffer) {
+ logger.trace("Server responded with message of type: {}",
value.getStatus());
+ switch (value.getStatus()) {
+ case SASL_AUTH_IN_PROGRESS: {
+ try {
+ final SaslMessage.Builder response = SaslMessage.newBuilder();
+ final byte[] responseBytes = evaluateChallenge(ugi, saslClient,
value.getData().toByteArray());
+ final boolean isComplete = saslClient.isComplete();
+ logger.trace("Evaluated challenge. Completed? {}. Sending
response to server.", isComplete);
+ response.setData(responseBytes != null ?
ByteString.copyFrom(responseBytes) : ByteString.EMPTY);
+ // if isComplete, the client will get one more response from
server
+ response.setStatus(isComplete ? SaslStatus.SASL_AUTH_SUCCESS :
SaslStatus.SASL_AUTH_IN_PROGRESS);
+ send(new SaslChallengeHandler(ugi, future),
+ connection,
+ RpcType.SASL_MESSAGE,
+ response.build(),
+ SaslMessage.class,
+ true // the connection will not be backed up at this point
+ );
+ } catch (Exception e) {
+ future.setException(e);
+ }
+ break;
+ }
+ case SASL_AUTH_SUCCESS: {
+ try {
+ if (saslClient.isComplete()) {
+ logger.trace("Successfully authenticated to server using {}",
saslClient.getMechanismName());
+ saslClient.dispose();
+ saslClient = null;
+ future.set(null); // success
+ } else {
+ // server completed before client; so try once, fail otherwise
+ evaluateChallenge(ugi, saslClient,
value.getData().toByteArray()); // discard response
+ if (saslClient.isComplete()) {
+ logger.trace("Successfully authenticated to server using
{}", saslClient.getMechanismName());
+ saslClient.dispose();
+ saslClient = null;
+ future.set(null); // success
+ } else {
+ future.setException(
+ new SaslException("Server allegedly succeeded
authentication, but client did not. Suspicious?"));
+ }
+ }
+ } catch (Exception e) {
+ future.setException(e);
+ }
+ break;
+ }
+ case SASL_AUTH_FAILED: {
+ future.setException(new SaslException("Incorrect credentials?"));
+ try {
+ saslClient.dispose();
+ } catch (final SaslException ignored) {
+ // ignored
+ }
+ saslClient = null;
+ break;
+ }
+ default:
+ future.setException(new SaslException("Server sent a corrupt
message."));
--- End diff --
Done.
> Kerberos Authentication
> -----------------------
>
> Key: DRILL-4280
> URL: https://issues.apache.org/jira/browse/DRILL-4280
> Project: Apache Drill
> Issue Type: Improvement
> Reporter: Keys Botzum
> Assignee: Chunhui Shi
> Labels: security
>
> Drill should support Kerberos based authentication from clients. This means
> that both the ODBC and JDBC drivers as well as the web/REST interfaces should
> support inbound Kerberos. For Web this would most likely be SPNEGO while for
> ODBC and JDBC this will be more generic Kerberos.
> Since Hive and much of Hadoop supports Kerberos there is a potential for a
> lot of reuse of ideas if not implementation.
> Note that this is related to but not the same as
> https://issues.apache.org/jira/browse/DRILL-3584
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
