You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by bo...@apache.org on 2016/05/27 23:31:25 UTC
hadoop git commit: HDFS-10366: libhdfs++: Add SASL authentication.
Contributed by Bob Hansen
Repository: hadoop
Updated Branches:
refs/heads/HDFS-8707 efc843e18 -> f0ef898ed
HDFS-10366: libhdfs++: Add SASL authentication. Contributed by Bob Hansen
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/f0ef898e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/f0ef898e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/f0ef898e
Branch: refs/heads/HDFS-8707
Commit: f0ef898edf24a4a3fb9262c0b609688e9323ed27
Parents: efc843e
Author: Bob Hansen <bo...@hp.com>
Authored: Fri May 27 19:29:24 2016 -0400
Committer: Bob Hansen <bo...@hp.com>
Committed: Fri May 27 19:29:24 2016 -0400
----------------------------------------------------------------------
.../native/libhdfspp/CMake/FindCyrusSASL.cmake | 49 +++
.../main/native/libhdfspp/CMake/FindGSasl.cmake | 44 +++
.../src/main/native/libhdfspp/CMakeLists.txt | 38 +++
.../main/native/libhdfspp/examples/cat/c/cat.c | 1 +
.../native/libhdfspp/include/hdfspp/options.h | 11 +
.../native/libhdfspp/include/hdfspp/status.h | 6 +-
.../native/libhdfspp/lib/common/CMakeLists.txt | 2 +-
.../native/libhdfspp/lib/common/auth_info.cc | 18 ++
.../native/libhdfspp/lib/common/auth_info.h | 90 ++++++
.../libhdfspp/lib/common/hdfs_configuration.cc | 9 +
.../libhdfspp/lib/common/hdfs_configuration.h | 4 +
.../main/native/libhdfspp/lib/common/options.cc | 3 +-
.../main/native/libhdfspp/lib/common/status.cc | 9 +-
.../native/libhdfspp/lib/rpc/CMakeLists.txt | 2 +-
.../native/libhdfspp/lib/rpc/rpc_connection.cc | 154 +++++++--
.../native/libhdfspp/lib/rpc/rpc_connection.h | 119 ++++---
.../main/native/libhdfspp/lib/rpc/rpc_engine.cc | 10 +-
.../main/native/libhdfspp/lib/rpc/rpc_engine.h | 47 ++-
.../native/libhdfspp/lib/rpc/sasl_engine.cc | 185 +++++++++++
.../main/native/libhdfspp/lib/rpc/sasl_engine.h | 125 ++++++++
.../native/libhdfspp/lib/rpc/sasl_protocol.cc | 320 +++++++++++++++++++
.../native/libhdfspp/lib/rpc/sasl_protocol.h | 81 +++++
.../main/native/libhdfspp/tests/CMakeLists.txt | 14 +-
23 files changed, 1238 insertions(+), 103 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMake/FindCyrusSASL.cmake
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMake/FindCyrusSASL.cmake b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMake/FindCyrusSASL.cmake
new file mode 100644
index 0000000..8ce027f
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMake/FindCyrusSASL.cmake
@@ -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
+#
+# 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.
+#
+# - Find Cyrus SASL (sasl.h, libsasl2.so)
+#
+# This module defines
+# CYRUS_SASL_INCLUDE_DIR, directory containing headers
+# CYRUS_SASL_SHARED_LIB, path to Cyrus SASL's shared library
+# CYRUS_SASL_FOUND, whether Cyrus SASL and its plugins have been found
+#
+# N.B: we do _not_ include sasl in thirdparty, for a fairly subtle reason. The
+# TLDR version is that newer versions of cyrus-sasl (>=2.1.26) have a bug fix
+# for https://bugzilla.cyrusimap.org/show_bug.cgi?id=3590, but that bug fix
+# relied on a change both on the plugin side and on the library side. If you
+# then try to run the new version of sasl (e.g from our thirdparty tree) with
+# an older version of a plugin (eg from RHEL6 install), you'll get a SASL_NOMECH
+# error due to this bug.
+#
+# In practice, Cyrus-SASL is so commonly used and generally non-ABI-breaking that
+# we should be OK to depend on the host installation.
+
+# Note that this is modified from the version that was copied from our
+# friends at the Kudu project. The original version implicitly required
+# the Cyrus SASL. This version will only complain if REQUIRED is added.
+
+
+find_path(CYRUS_SASL_INCLUDE_DIR sasl/sasl.h)
+find_library(CYRUS_SASL_SHARED_LIB sasl2)
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(CYRUS_SASL DEFAULT_MSG
+ CYRUS_SASL_SHARED_LIB CYRUS_SASL_INCLUDE_DIR)
+
+MARK_AS_ADVANCED(CYRUS_SASL_INCLUDE_DIR CYRUS_SASL_SHARED_LIB)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMake/FindGSasl.cmake
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMake/FindGSasl.cmake b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMake/FindGSasl.cmake
new file mode 100644
index 0000000..57588ad
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMake/FindGSasl.cmake
@@ -0,0 +1,44 @@
+#
+# 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.
+#
+
+# - Try to find the GNU sasl library (gsasl)
+#
+# Once done this will define
+#
+# GSASL_FOUND - System has gnutls
+# GSASL_INCLUDE_DIR - The gnutls include directory
+# GSASL_LIBRARIES - The libraries needed to use gnutls
+# GSASL_DEFINITIONS - Compiler switches required for using gnutls
+
+
+IF (GSASL_INCLUDE_DIR AND GSASL_LIBRARIES)
+ # in cache already
+ SET(GSasl_FIND_QUIETLY TRUE)
+ENDIF (GSASL_INCLUDE_DIR AND GSASL_LIBRARIES)
+
+FIND_PATH(GSASL_INCLUDE_DIR gsasl.h)
+
+FIND_LIBRARY(GSASL_LIBRARIES gsasl)
+
+INCLUDE(FindPackageHandleStandardArgs)
+
+# handle the QUIETLY and REQUIRED arguments and set GSASL_FOUND to TRUE if
+# all listed variables are TRUE
+FIND_PACKAGE_HANDLE_STANDARD_ARGS(GSASL DEFAULT_MSG GSASL_LIBRARIES GSASL_INCLUDE_DIR)
+
+MARK_AS_ADVANCED(GSASL_INCLUDE_DIR GSASL_LIBRARIES)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMakeLists.txt
index 77f63c0..a663e60 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMakeLists.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMakeLists.txt
@@ -22,10 +22,14 @@ cmake_minimum_required(VERSION 2.8)
enable_testing()
include (CTest)
+SET(CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/CMake" ${CMAKE_MODULE_PATH})
+SET(CMAKE_DISABLE_FIND_PACKAGE_CyrusSASL TRUE) # Until development is done.
find_package(Doxygen)
find_package(OpenSSL REQUIRED)
find_package(Protobuf REQUIRED)
+find_package(GSasl)
+find_package(CyrusSASL)
find_package(Threads)
find_program(MEMORYCHECK_COMMAND valgrind HINTS ${VALGRIND_DIR} )
@@ -37,6 +41,36 @@ if (REQUIRE_VALGRIND AND MEMORYCHECK_COMMAND MATCHES "MEMORYCHECK_COMMAND-NOTFOU
"The path can be included via a -DVALGRIND_DIR=... flag passed to CMake.")
endif (REQUIRE_VALGRIND AND MEMORYCHECK_COMMAND MATCHES "MEMORYCHECK_COMMAND-NOTFOUND" )
+# Prefer Cyrus SASL, but use GSASL if it is found
+# Note that the packages can be disabled by setting CMAKE_DISABLE_FIND_PACKAGE_GSasl or
+# CMAKE_DISABLE_FIND_PACKAGE_CyrusSASL, respectively (case sensitive)
+set (SASL_LIBRARIES)
+set (SASL_INCLUDE_DIR)
+if (CYRUS_SASL_FOUND)
+ message(STATUS "Using Cyrus SASL; link with ${CYRUS_SASL_LIBRARIES}")
+ set (SASL_INCLUDE_DIR ${CYRUS_SASL_INCLUDE_DIR})
+ set (SASL_LIBRARIES ${CYRUS_SASL_SHARED_LIB})
+ add_definitions(-DUSE_SASL -DUSE_CYRUS_SASL)
+else (CYRUS_SASL_FOUND)
+ if (REQUIRE_CYRUS_SASL)
+ message(FATAL_ERROR "Cyrus SASL was required but not found. "
+ "The path can be included via a -DCYRUS_SASL_DIR=... flag passed to CMake.")
+ endif (REQUIRE_CYRUS_SASL)
+
+ # If we didn't pick Cyrus, use GSASL instead
+ if (GSASL_FOUND)
+ message(STATUS "Using GSASL; link with ${GSASL_LIBRARIES}")
+ set (SASL_INCLUDE_DIR ${GSASL_INCLUDE_DIR})
+ set (SASL_LIBRARIES ${GSASL_LIBRARIES})
+ add_definitions(-DUSE_SASL -DUSE_GSASL)
+ else (GSASL_FOUND)
+ if (REQUIRE_GSASL)
+ message(FATAL_ERROR "GSASL was required but not found. "
+ "The path can be included via a -DGSASL_DIR=... flag passed to CMake.")
+ endif (REQUIRE_GSASL)
+ message(STATUS "Not using SASL")
+ endif (GSASL_FOUND)
+endif (CYRUS_SASL_FOUND)
add_definitions(-DASIO_STANDALONE -DASIO_CPP11_DATE_TIME)
@@ -120,6 +154,7 @@ include_directories( SYSTEM
third_party/protobuf
third_party/uriparser2
${OPENSSL_INCLUDE_DIR}
+ ${SASL_INCLUDE_DIR}
)
@@ -146,6 +181,7 @@ if (HADOOP_BUILD)
${LIB_DL}
${PROTOBUF_LIBRARY}
${OPENSSL_LIBRARIES}
+ ${SASL_LIBRARIES}
)
else (HADOOP_BUILD)
add_library(hdfspp_static STATIC ${EMPTY_FILE_CC} ${LIBHDFSPP_ALL_OBJECTS})
@@ -153,12 +189,14 @@ else (HADOOP_BUILD)
${LIB_DL}
${PROTOBUF_LIBRARY}
${OPENSSL_LIBRARIES}
+ ${SASL_LIBRARIES}
)
add_library(hdfspp SHARED ${EMPTY_FILE_CC} ${LIBHDFSPP_ALL_OBJECTS})
target_link_libraries(hdfspp_static
${LIB_DL}
${PROTOBUF_LIBRARY}
${OPENSSL_LIBRARIES}
+ ${SASL_LIBRARIES}
)
endif (HADOOP_BUILD)
set(LIBHDFSPP_VERSION "0.1.0")
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cat/c/cat.c
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cat/c/cat.c b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cat/c/cat.c
index b5fe311..5839308 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cat/c/cat.c
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cat/c/cat.c
@@ -66,6 +66,7 @@ void parse_uri(const char * uri_string, struct Uri * uri) {
char * host_port_separator = strstr(authority, ":");
if (host_port_separator != NULL) {
+ errno = 0;
uri->port = strtol(host_port_separator + 1, NULL, 10);
if (errno != 0)
return;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/options.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/options.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/options.h
index f63e70b..1828b2a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/options.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/options.h
@@ -58,6 +58,17 @@ struct Options {
*/
URI defaultFS;
+ /**
+ * Which form of authentication to use with the server
+ * Default: simple
+ */
+ enum Authentication {
+ kSimple,
+ kKerberos
+ };
+ Authentication authentication;
+ static const Authentication kDefaultAuthentication = kSimple;
+
Options();
};
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/status.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/status.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/status.h
index 3c7563d..6a3ce73 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/status.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/status.h
@@ -37,7 +37,8 @@ class Status {
static Status Unimplemented();
static Status Exception(const char *expception_class_name, const char *error_message);
static Status Error(const char *error_message);
- static Status Canceled();
+ static Status AuthenticationFailed();
+ static Status Canceled();
// success
bool ok() const { return code_ == 0; }
@@ -55,7 +56,8 @@ class Status {
kUnimplemented = static_cast<unsigned>(std::errc::function_not_supported),
kOperationCanceled = static_cast<unsigned>(std::errc::operation_canceled),
kPermissionDenied = static_cast<unsigned>(std::errc::permission_denied),
- kException = 255,
+ kException = 256,
+ kAuthenticationFailed = 257,
};
private:
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/CMakeLists.txt
index ea2f952..484d1a4 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/CMakeLists.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/CMakeLists.txt
@@ -19,6 +19,6 @@ if(NEED_LINK_DL)
set(LIB_DL dl)
endif()
-add_library(common_obj OBJECT base64.cc status.cc sasl_digest_md5.cc hdfs_public_api.cc options.cc configuration.cc configuration_loader.cc hdfs_configuration.cc uri.cc util.cc retry_policy.cc cancel_tracker.cc logging.cc libhdfs_events_impl.cc)
+add_library(common_obj OBJECT base64.cc status.cc sasl_digest_md5.cc hdfs_public_api.cc options.cc configuration.cc configuration_loader.cc hdfs_configuration.cc uri.cc util.cc retry_policy.cc cancel_tracker.cc logging.cc libhdfs_events_impl.cc auth_info.cc)
add_library(common $<TARGET_OBJECTS:common_obj> $<TARGET_OBJECTS:uriparser2_obj>)
target_link_libraries(common ${LIB_DL})
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/auth_info.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/auth_info.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/auth_info.cc
new file mode 100644
index 0000000..fb98329
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/auth_info.cc
@@ -0,0 +1,18 @@
+/**
+ * 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.
+ */
+#include "auth_info.h"
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/auth_info.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/auth_info.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/auth_info.h
new file mode 100644
index 0000000..3f2f1af
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/auth_info.h
@@ -0,0 +1,90 @@
+/**
+ * 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.
+ */
+
+#ifndef LIB_FS_AUTHINFO_H
+#define LIB_FS_AUTHINFO_H
+
+#include <optional.hpp>
+
+namespace hdfs {
+
+class Token {
+public:
+ std::string identifier;
+ std::string password;
+};
+
+class AuthInfo {
+public:
+ enum AuthMethod {
+ kSimple,
+ kKerberos,
+ kToken,
+ kUnknownAuth,
+ kAuthFailed
+ };
+
+ AuthInfo() :
+ method(kSimple) {
+ }
+
+ explicit AuthInfo(AuthMethod mech) :
+ method(mech) {
+ }
+
+ bool useSASL() {
+ return method != kSimple;
+ }
+
+ const std::string & getUser() const {
+ return user;
+ }
+
+ void setUser(const std::string & user) {
+ this->user = user;
+ }
+
+ AuthMethod getMethod() const {
+ return method;
+ }
+
+ void setMethod(AuthMethod method) {
+ this->method = method;
+ }
+
+ const std::experimental::optional<Token> & getToken() const {
+ return token;
+ }
+
+ void setToken(const Token & token) {
+ this->token = token;
+ }
+
+ void clearToken() {
+ this->token = std::experimental::nullopt;
+ }
+
+private:
+ AuthMethod method;
+ std::string user;
+ std::experimental::optional<Token> token;
+};
+
+}
+
+#endif /* RPCAUTHINFO_H */
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/hdfs_configuration.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/hdfs_configuration.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/hdfs_configuration.cc
index 6d50a29..13ba279 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/hdfs_configuration.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/hdfs_configuration.cc
@@ -49,6 +49,15 @@ Options HdfsConfiguration::GetOptions() {
OptionalSet(result.rpc_retry_delay_ms, GetInt(kIpcClientConnectRetryIntervalKey));
OptionalSet(result.defaultFS, GetUri(kFsDefaultFsKey));
+ optional<std::string> authentication_value = Get(kHadoopSecurityAuthentication);
+ if (authentication_value ) {
+ std::string fixed_case_value = fixCase(authentication_value.value());
+ if (fixed_case_value == fixCase(kHadoopSecurityAuthentication_kerberos))
+ result.authentication = Options::kKerberos;
+ else
+ result.authentication = Options::kSimple;
+ }
+
return result;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/hdfs_configuration.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/hdfs_configuration.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/hdfs_configuration.h
index 5a39b11..7db9d37 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/hdfs_configuration.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/hdfs_configuration.h
@@ -41,6 +41,10 @@ class HdfsConfiguration : public Configuration {
static constexpr const char * kDfsClientSocketTimeoutKey = "dfs.client.socket-timeout";
static constexpr const char * kIpcClientConnectMaxRetriesKey = "ipc.client.connect.max.retries";
static constexpr const char * kIpcClientConnectRetryIntervalKey = "ipc.client.connect.retry.interval";
+ static constexpr const char * kHadoopSecurityAuthentication = "hadoop.security.authentication";
+ static constexpr const char * kHadoopSecurityAuthentication_simple = "simple";
+ static constexpr const char * kHadoopSecurityAuthentication_kerberos = "kerberos";
+
private:
friend class ConfigurationLoader;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/options.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/options.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/options.cc
index 6406dea..c7dd2ed 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/options.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/options.cc
@@ -30,6 +30,7 @@ const unsigned int Options::kDefaultHostExclusionDuration;
Options::Options() : rpc_timeout(kDefaultRpcTimeout), max_rpc_retries(kDefaultMaxRpcRetries),
rpc_retry_delay_ms(kDefaultRpcRetryDelayMs),
host_exclusion_duration(kDefaultHostExclusionDuration),
- defaultFS()
+ defaultFS(),
+ authentication(kDefaultAuthentication)
{}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/status.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/status.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/status.cc
index 78d4379..e215a67 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/status.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/status.cc
@@ -25,6 +25,7 @@
namespace hdfs {
const char * kStatusAccessControlException = "org.apache.hadoop.security.AccessControlException";
+const char * kStatusSaslException = "javax.security.sasl.SaslException";
Status::Status(int code, const char *msg1) : code_(code) {
if(msg1) {
@@ -63,12 +64,18 @@ Status Status::Unimplemented() {
Status Status::Exception(const char *exception_class_name, const char *error_message) {
if (exception_class_name && (strcmp(exception_class_name, kStatusAccessControlException) == 0) )
return Status(kPermissionDenied, error_message);
+ else if (exception_class_name && (strcmp(exception_class_name, kStatusSaslException) == 0))
+ return AuthenticationFailed();
else
return Status(kException, exception_class_name, error_message);
}
Status Status::Error(const char *error_message) {
- return Exception("Exception", error_message);
+ return Status(kAuthenticationFailed, error_message);
+}
+
+Status Status::AuthenticationFailed() {
+ return Status(kAuthenticationFailed, "Authentication failed");
}
Status Status::Canceled() {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/CMakeLists.txt
index b82bcf6..8c80633 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/CMakeLists.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/CMakeLists.txt
@@ -17,6 +17,6 @@
#
include_directories(${OPENSSL_INCLUDE_DIRS})
-add_library(rpc_obj OBJECT rpc_connection.cc rpc_engine.cc)
+add_library(rpc_obj OBJECT rpc_connection.cc rpc_engine.cc sasl_protocol.cc sasl_engine.cc)
add_dependencies(rpc_obj proto)
add_library(rpc $<TARGET_OBJECTS:rpc_obj>)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.cc
index 6c3b82e..749195a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.cc
@@ -16,6 +16,7 @@
* limitations under the License.
*/
#include "rpc_engine.h"
+#include "sasl_protocol.h"
#include "RpcHeader.pb.h"
#include "ProtobufRpcEngine.pb.h"
@@ -110,22 +111,22 @@ static void SetRequestHeader(LockFreeRpcEngine *engine, int call_id,
RpcConnection::~RpcConnection() {}
-Request::Request(LockFreeRpcEngine *engine, const std::string &method_name,
+Request::Request(LockFreeRpcEngine *engine, const std::string &method_name, int call_id,
const std::string &request, Handler &&handler)
: engine_(engine),
method_name_(method_name),
- call_id_(engine->NextCallId()),
+ call_id_(call_id),
timer_(engine->io_service()),
handler_(std::move(handler)),
retry_count_(engine->retry_policy() ? 0 : kNoRetry) {
ConstructPayload(&payload_, &request);
}
-Request::Request(LockFreeRpcEngine *engine, const std::string &method_name,
+Request::Request(LockFreeRpcEngine *engine, const std::string &method_name, int call_id,
const pb::MessageLite *request, Handler &&handler)
: engine_(engine),
method_name_(method_name),
- call_id_(engine->NextCallId()),
+ call_id_(call_id),
timer_(engine->io_service()),
handler_(std::move(handler)),
retry_count_(engine->retry_policy() ? 0 : kNoRetry) {
@@ -148,7 +149,12 @@ void Request::GetPacket(std::string *res) const {
RequestHeaderProto req_header;
SetRequestHeader(engine_, call_id_, method_name_, retry_count_, &rpc_header,
&req_header);
- AddHeadersToPacket(res, {&rpc_header, &req_header}, &payload_);
+
+ // SASL messages don't have a request header
+ if (method_name_ != SASL_METHOD_NAME)
+ AddHeadersToPacket(res, {&rpc_header, &req_header}, &payload_);
+ else
+ AddHeadersToPacket(res, {&rpc_header}, &payload_);
}
void Request::OnResponseArrived(pbio::CodedInputStream *is,
@@ -171,12 +177,80 @@ void RpcConnection::StartReading() {
});
}
+void RpcConnection::HandshakeComplete(const Status &s) {
+ std::lock_guard<std::mutex> state_lock(connection_state_lock_);
+
+ LOG_TRACE(kRPC, << "RpcConnectionImpl::HandshakeComplete called");
+
+ if (s.ok()) {
+ if (connected_ == kConnecting) {
+ auto shared_this = shared_from_this();
+
+ connected_ = kAuthenticating;
+ if (auth_info_.useSASL()) {
+#ifdef USE_SASL
+ sasl_protocol_ = std::make_shared<SaslProtocol>(cluster_name_, auth_info_, shared_from_this());
+ sasl_protocol_->SetEventHandlers(event_handlers_);
+ sasl_protocol_->authenticate([shared_this, this](
+ const Status & status, const AuthInfo & new_auth_info) {
+ AuthComplete(status, new_auth_info); } );
+#else
+ AuthComplete_locked(Status::Error("SASL is required, but no SASL library was found"), auth_info_);
+#endif
+ } else {
+ AuthComplete_locked(Status::OK(), auth_info_);
+ }
+ }
+ } else {
+ CommsError(s);
+ };
+}
+
+void RpcConnection::AuthComplete(const Status &s, const AuthInfo & new_auth_info) {
+ std::lock_guard<std::mutex> state_lock(connection_state_lock_);
+ AuthComplete_locked(s, new_auth_info);
+}
+
+void RpcConnection::AuthComplete_locked(const Status &s, const AuthInfo & new_auth_info) {
+ assert(lock_held(connection_state_lock_)); // Must be holding lock before calling
+ LOG_TRACE(kRPC, << "RpcConnectionImpl::AuthComplete called");
+
+ // Free the sasl_protocol object
+ sasl_protocol_.reset();
+
+ if (s.ok()) {
+ auth_info_ = new_auth_info;
+
+ auto shared_this = shared_from_this();
+ SendContext([shared_this, this](const Status & s) {
+ ContextComplete(s);
+ });
+ } else {
+ CommsError(s);
+ };
+}
+
+void RpcConnection::ContextComplete(const Status &s) {
+ std::lock_guard<std::mutex> state_lock(connection_state_lock_);
+
+ LOG_TRACE(kRPC, << "RpcConnectionImpl::ContextComplete called");
+
+ if (s.ok()) {
+ if (connected_ == kAuthenticating) {
+ connected_ = kConnected;
+ }
+ FlushPendingRequests();
+ } else {
+ CommsError(s);
+ };
+}
+
void RpcConnection::AsyncFlushPendingRequests() {
std::shared_ptr<RpcConnection> shared_this = shared_from_this();
io_service().post([shared_this, this]() {
std::lock_guard<std::mutex> state_lock(connection_state_lock_);
- LOG_TRACE(kRPC, << "RpcConnection::AsyncRpc called (connected=" << ToString(connected_) << ")");
+ LOG_TRACE(kRPC, << "RpcConnection::AsyncFlushPendingRequests called (connected=" << ToString(connected_) << ")");
if (!request_over_the_wire_) {
FlushPendingRequests();
@@ -246,10 +320,22 @@ std::shared_ptr<std::string> RpcConnection::PrepareHandshakePacket() {
*
* AuthProtocol: 0->none, -33->SASL
*/
- static const char kHandshakeHeader[] = {'h', 'r', 'p', 'c',
- RpcEngine::kRpcVersion, 0, 0};
+
+ char auth_protocol = auth_info_.useSASL() ? -33 : 0;
+ const char handshake_header[] = {'h', 'r', 'p', 'c',
+ RpcEngine::kRpcVersion, 0, auth_protocol};
auto res =
- std::make_shared<std::string>(kHandshakeHeader, sizeof(kHandshakeHeader));
+ std::make_shared<std::string>(handshake_header, sizeof(handshake_header));
+
+ return res;
+}
+
+std::shared_ptr<std::string> RpcConnection::PrepareContextPacket() {
+ // This needs to be send after the SASL handshake, and
+ // after the SASL handshake (if any)
+ assert(lock_held(connection_state_lock_)); // Must be holding lock before calling
+
+ auto res = std::make_shared<std::string>();
RpcRequestHeaderProto h;
h.set_rpckind(RPC_PROTOCOL_BUFFER);
@@ -259,11 +345,12 @@ std::shared_ptr<std::string> RpcConnection::PrepareHandshakePacket() {
IpcConnectionContextProto handshake;
handshake.set_protocol(engine_->protocol_name());
- const std::string & user_name = engine()->user_name();
+ const std::string & user_name = auth_info_.getUser();
if (!user_name.empty()) {
*handshake.mutable_userinfo()->mutable_effectiveuser() = user_name;
}
AddHeadersToPacket(res.get(), {&h, &handshake}, nullptr);
+
return res;
}
@@ -272,6 +359,14 @@ void RpcConnection::AsyncRpc(
std::shared_ptr<::google::protobuf::MessageLite> resp,
const RpcCallback &handler) {
std::lock_guard<std::mutex> state_lock(connection_state_lock_);
+ AsyncRpc_locked(method_name, req, resp, handler);
+}
+
+void RpcConnection::AsyncRpc_locked(
+ const std::string &method_name, const ::google::protobuf::MessageLite *req,
+ std::shared_ptr<::google::protobuf::MessageLite> resp,
+ const RpcCallback &handler) {
+ assert(lock_held(connection_state_lock_)); // Must be holding lock before calling
auto wrapped_handler =
[resp, handler](pbio::CodedInputStream *is, const Status &status) {
@@ -283,29 +378,21 @@ void RpcConnection::AsyncRpc(
handler(status);
};
- auto r = std::make_shared<Request>(engine_, method_name, req,
+ int call_id = (method_name != SASL_METHOD_NAME ? engine_->NextCallId() : RpcEngine::kCallIdSasl);
+ auto r = std::make_shared<Request>(engine_, method_name, call_id, req,
std::move(wrapped_handler));
-
- if (connected_ == kDisconnected) {
- // Oops. The connection failed _just_ before the engine got a chance
- // to send it. Register it as a failure
- Status status = Status::ResourceUnavailable("RpcConnection closed before send.");
- auto r_vector = std::vector<std::shared_ptr<Request> > (1, r);
- assert(r_vector[0].get() != nullptr);
-
- engine_->AsyncRpcCommsError(status, shared_from_this(), r_vector);
- } else {
- pending_requests_.push_back(r);
-
- if (connected_ == kConnected) { // Dont flush if we're waiting or handshaking
- FlushPendingRequests();
- }
- }
+ auto r_vector = std::vector<std::shared_ptr<Request> > (1, r);
+ SendRpcRequests(r_vector);
}
void RpcConnection::AsyncRpc(const std::vector<std::shared_ptr<Request> > & requests) {
std::lock_guard<std::mutex> state_lock(connection_state_lock_);
- LOG_TRACE(kRPC, << "RpcConnection::AsyncRpc[] called; connected=" << ToString(connected_));
+ SendRpcRequests(requests);
+}
+
+void RpcConnection::SendRpcRequests(const std::vector<std::shared_ptr<Request> > & requests) {
+ LOG_TRACE(kRPC, << "RpcConnection::SendRpcRequests[] called; connected=" << ToString(connected_));
+ assert(lock_held(connection_state_lock_)); // Must be holding lock before calling
if (connected_ == kDisconnected) {
// Oops. The connection failed _just_ before the engine got a chance
@@ -315,9 +402,12 @@ void RpcConnection::AsyncRpc(const std::vector<std::shared_ptr<Request> > & requ
} else {
pending_requests_.reserve(pending_requests_.size() + requests.size());
for (auto r: requests) {
- pending_requests_.push_back(r);
+ if (r->method_name() != SASL_METHOD_NAME)
+ pending_requests_.push_back(r);
+ else
+ auth_requests_.push_back(r);
}
- if (connected_ == kConnected) { // Dont flush if we're waiting or handshaking
+ if (connected_ == kConnected || connected_ == kAuthenticating) { // Dont flush if we're waiting or handshaking
FlushPendingRequests();
}
}
@@ -341,6 +431,9 @@ void RpcConnection::PreEnqueueRequests(
void RpcConnection::SetEventHandlers(std::shared_ptr<LibhdfsEvents> event_handlers) {
std::lock_guard<std::mutex> state_lock(connection_state_lock_);
event_handlers_ = event_handlers;
+ if (sasl_protocol_) {
+ sasl_protocol_->SetEventHandlers(event_handlers);
+ }
}
void RpcConnection::SetClusterName(std::string cluster_name) {
@@ -401,6 +494,7 @@ std::string RpcConnection::ToString(ConnectedState connected) {
switch(connected) {
case kNotYetConnected: return "NotYetConnected";
case kConnecting: return "Connecting";
+ case kAuthenticating: return "Authenticating";
case kConnected: return "Connected";
case kDisconnected: return "Disconnected";
default: return "Invalid ConnectedState";
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h
index d7b1c2f..255b98b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h
@@ -20,9 +20,11 @@
#include "rpc_engine.h"
+#include "common/auth_info.h"
#include "common/logging.h"
#include "common/util.h"
#include "common/libhdfs_events_impl.h"
+#include "sasl_protocol.h"
#include <asio/connect.hpp>
#include <asio/read.hpp>
@@ -37,10 +39,12 @@ public:
virtual ~RpcConnectionImpl() override;
virtual void Connect(const std::vector<::asio::ip::tcp::endpoint> &server,
+ const AuthInfo & auth_info,
RpcCallback &handler);
virtual void ConnectAndFlush(
const std::vector<::asio::ip::tcp::endpoint> &server) override;
- virtual void Handshake(RpcCallback &handler) override;
+ virtual void SendHandshake(RpcCallback &handler) override;
+ virtual void SendContext(RpcCallback &handler) override;
virtual void Disconnect() override;
virtual void OnSendCompleted(const ::asio::error_code &ec,
size_t transferred) override;
@@ -59,7 +63,6 @@ public:
NextLayer next_layer_;
void ConnectComplete(const ::asio::error_code &ec);
- void HandshakeComplete(const Status &s);
};
template <class NextLayer>
@@ -84,8 +87,13 @@ RpcConnectionImpl<NextLayer>::~RpcConnectionImpl() {
template <class NextLayer>
void RpcConnectionImpl<NextLayer>::Connect(
- const std::vector<::asio::ip::tcp::endpoint> &server, RpcCallback &handler) {
+ const std::vector<::asio::ip::tcp::endpoint> &server,
+ const AuthInfo & auth_info,
+ RpcCallback &handler) {
LOG_TRACE(kRPC, << "RpcConnectionImpl::Connect called");
+
+ this->auth_info_ = auth_info;
+
auto connectionSuccessfulReq = std::make_shared<Request>(
engine_, [handler](::google::protobuf::io::CodedInputStream *is,
const Status &status) {
@@ -147,7 +155,7 @@ void RpcConnectionImpl<NextLayer>::ConnectComplete(const ::asio::error_code &ec)
if (status.ok()) {
StartReading();
- Handshake([shared_this, this](const Status & s) {
+ SendHandshake([shared_this, this](const Status & s) {
HandshakeComplete(s);
});
} else {
@@ -172,34 +180,33 @@ void RpcConnectionImpl<NextLayer>::ConnectComplete(const ::asio::error_code &ec)
}
template <class NextLayer>
-void RpcConnectionImpl<NextLayer>::HandshakeComplete(const Status &s) {
- std::lock_guard<std::mutex> state_lock(connection_state_lock_);
+void RpcConnectionImpl<NextLayer>::SendHandshake(RpcCallback &handler) {
+ assert(lock_held(connection_state_lock_)); // Must be holding lock before calling
- LOG_TRACE(kRPC, << "RpcConnectionImpl::HandshakeComplete called");
+ LOG_TRACE(kRPC, << "RpcConnectionImpl::SendHandshake called");
- if (s.ok()) {
- FlushPendingRequests();
- } else {
- CommsError(s);
- };
+ auto shared_this = shared_from_this();
+ auto handshake_packet = PrepareHandshakePacket();
+ ::asio::async_write(next_layer_, asio::buffer(*handshake_packet),
+ [handshake_packet, handler, shared_this, this](
+ const ::asio::error_code &ec, size_t) {
+ Status status = ToStatus(ec);
+ handler(status);
+ });
}
-
template <class NextLayer>
-void RpcConnectionImpl<NextLayer>::Handshake(RpcCallback &handler) {
+void RpcConnectionImpl<NextLayer>::SendContext(RpcCallback &handler) {
assert(lock_held(connection_state_lock_)); // Must be holding lock before calling
- LOG_TRACE(kRPC, << "RpcConnectionImpl::Handshake called");
+ LOG_TRACE(kRPC, << "RpcConnectionImpl::SendContext called");
auto shared_this = shared_from_this();
- auto handshake_packet = PrepareHandshakePacket();
- ::asio::async_write(next_layer_, asio::buffer(*handshake_packet),
- [handshake_packet, handler, shared_this, this](
+ auto context_packet = PrepareContextPacket();
+ ::asio::async_write(next_layer_, asio::buffer(*context_packet),
+ [context_packet, handler, shared_this, this](
const ::asio::error_code &ec, size_t) {
Status status = ToStatus(ec);
- if (status.ok() && connected_ == kConnecting) {
- connected_ = kConnected;
- }
handler(status);
});
}
@@ -232,29 +239,43 @@ void RpcConnectionImpl<NextLayer>::FlushPendingRequests() {
LOG_TRACE(kRPC, << "RpcConnectionImpl::FlushPendingRequests called");
- if (pending_requests_.empty()) {
+ // Don't send if we don't need to
+ if (request_over_the_wire_) {
return;
}
- if (connected_ == kDisconnected) {
- LOG_WARN(kRPC, << "RpcConnectionImpl::FlushPendingRequests attempted to flush a disconnected connection");
+ std::shared_ptr<Request> req;
+ switch (connected_) {
+ case kNotYetConnected:
return;
- }
- if (connected_ != kConnected) {
+ case kConnecting:
+ return;
+ case kAuthenticating:
+ if (auth_requests_.empty()) {
+ return;
+ }
+ req = auth_requests_.front();
+ auth_requests_.erase(auth_requests_.begin());
+ break;
+ case kConnected:
+ if (pending_requests_.empty()) {
+ return;
+ }
+ req = pending_requests_.front();
+ pending_requests_.erase(pending_requests_.begin());
+ break;
+ case kDisconnected:
LOG_DEBUG(kRPC, << "RpcConnectionImpl::FlushPendingRequests attempted to flush a " << ToString(connected_) << " connection");
- }
-
- // Don't send if we don't need to
- if (request_over_the_wire_) {
+ return;
+ default:
+ LOG_DEBUG(kRPC, << "RpcConnectionImpl::FlushPendingRequests invalid state: " << ToString(connected_));
return;
}
- std::shared_ptr<Request> req = pending_requests_.front();
- auto weak_req = std::weak_ptr<Request>(req);
- pending_requests_.erase(pending_requests_.begin());
-
std::shared_ptr<RpcConnection> shared_this = shared_from_this();
auto weak_this = std::weak_ptr<RpcConnection>(shared_this);
+ auto weak_req = std::weak_ptr<Request>(req);
+
std::shared_ptr<std::string> payload = std::make_shared<std::string>();
req->GetPacket(payload.get());
if (!payload->empty()) {
@@ -322,31 +343,31 @@ void RpcConnectionImpl<NextLayer>::OnRecvCompleted(const ::asio::error_code &asi
return;
}
- if (!response_) { /* start a new one */
- response_ = std::make_shared<Response>();
+ if (!current_response_state_) { /* start a new one */
+ current_response_state_ = std::make_shared<Response>();
}
- if (response_->state_ == Response::kReadLength) {
- response_->state_ = Response::kReadContent;
- auto buf = ::asio::buffer(reinterpret_cast<char *>(&response_->length_),
- sizeof(response_->length_));
+ if (current_response_state_->state_ == Response::kReadLength) {
+ current_response_state_->state_ = Response::kReadContent;
+ auto buf = ::asio::buffer(reinterpret_cast<char *>(¤t_response_state_->length_),
+ sizeof(current_response_state_->length_));
asio::async_read(
next_layer_, buf,
[shared_this, this](const ::asio::error_code &ec, size_t size) {
OnRecvCompleted(ec, size);
});
- } else if (response_->state_ == Response::kReadContent) {
- response_->state_ = Response::kParseResponse;
- response_->length_ = ntohl(response_->length_);
- response_->data_.resize(response_->length_);
+ } else if (current_response_state_->state_ == Response::kReadContent) {
+ current_response_state_->state_ = Response::kParseResponse;
+ current_response_state_->length_ = ntohl(current_response_state_->length_);
+ current_response_state_->data_.resize(current_response_state_->length_);
asio::async_read(
- next_layer_, ::asio::buffer(response_->data_),
+ next_layer_, ::asio::buffer(current_response_state_->data_),
[shared_this, this](const ::asio::error_code &ec, size_t size) {
OnRecvCompleted(ec, size);
});
- } else if (response_->state_ == Response::kParseResponse) {
- HandleRpcResponse(response_);
- response_ = nullptr;
+ } else if (current_response_state_->state_ == Response::kParseResponse) {
+ HandleRpcResponse(current_response_state_);
+ current_response_state_ = nullptr;
StartReading();
}
}
@@ -358,7 +379,7 @@ void RpcConnectionImpl<NextLayer>::Disconnect() {
LOG_INFO(kRPC, << "RpcConnectionImpl::Disconnect called");
request_over_the_wire_.reset();
- if (connected_ == kConnecting || connected_ == kConnected) {
+ if (connected_ == kConnecting || connected_ == kAuthenticating || connected_ == kConnected) {
// Don't print out errors, we were expecting a disconnect here
SafeDisconnect(get_asio_socket_ptr(&next_layer_));
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.cc
index 132bb69..f8df97f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.cc
@@ -34,13 +34,18 @@ RpcEngine::RpcEngine(::asio::io_service *io_service, const Options &options,
: io_service_(io_service),
options_(options),
client_name_(client_name),
- user_name_(user_name),
protocol_name_(protocol_name),
protocol_version_(protocol_version),
retry_policy_(std::move(MakeRetryPolicy(options))),
call_id_(0),
retry_timer(*io_service),
event_handlers_(std::make_shared<LibhdfsEvents>()) {
+
+ auth_info_.setUser(user_name);
+ if (options.authentication == Options::kKerberos) {
+ auth_info_.setMethod(AuthInfo::kKerberos);
+ }
+
LOG_DEBUG(kRPC, << "RpcEngine::RpcEngine called");
}
@@ -54,7 +59,7 @@ void RpcEngine::Connect(const std::string &cluster_name,
cluster_name_ = cluster_name;
conn_ = InitializeConnection();
- conn_->Connect(last_endpoints_, handler);
+ conn_->Connect(last_endpoints_, auth_info_, handler);
}
void RpcEngine::Shutdown() {
@@ -121,6 +126,7 @@ std::shared_ptr<RpcConnection> RpcEngine::InitializeConnection()
return result;
}
+
void RpcEngine::AsyncRpcCommsError(
const Status &status,
std::shared_ptr<RpcConnection> failedConnection,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h
index b35a663..066c01f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h
@@ -21,6 +21,7 @@
#include "hdfspp/options.h"
#include "hdfspp/status.h"
+#include "common/auth_info.h"
#include "common/retry_policy.h"
#include "common/libhdfs_events_impl.h"
#include "common/new_delete.h"
@@ -56,6 +57,7 @@ typedef const std::function<void(const Status &)> RpcCallback;
class LockFreeRpcEngine;
class RpcConnection;
+class SaslProtocol;
/*
* Internal bookkeeping for an outstanding request from the consumer.
@@ -69,9 +71,9 @@ class Request {
typedef std::function<void(::google::protobuf::io::CodedInputStream *is,
const Status &status)> Handler;
- Request(LockFreeRpcEngine *engine, const std::string &method_name,
+ Request(LockFreeRpcEngine *engine, const std::string &method_name, int call_id,
const std::string &request, Handler &&callback);
- Request(LockFreeRpcEngine *engine, const std::string &method_name,
+ Request(LockFreeRpcEngine *engine, const std::string &method_name, int call_id,
const ::google::protobuf::MessageLite *request, Handler &&callback);
// Null request (with no actual message) used to track the state of an
@@ -79,6 +81,7 @@ class Request {
Request(LockFreeRpcEngine *engine, Handler &&handler);
int call_id() const { return call_id_; }
+ std::string method_name() const { return method_name_; }
::asio::deadline_timer &timer() { return timer_; }
int IncrementRetryCount() { return retry_count_++; }
void GetPacket(std::string *res) const;
@@ -117,9 +120,9 @@ class RpcConnection : public std::enable_shared_from_this<RpcConnection> {
// Note that a single server can have multiple endpoints - especially both
// an ipv4 and ipv6 endpoint
virtual void Connect(const std::vector<::asio::ip::tcp::endpoint> &server,
+ const AuthInfo & auth_info,
RpcCallback &handler) = 0;
virtual void ConnectAndFlush(const std::vector<::asio::ip::tcp::endpoint> &server) = 0;
- virtual void Handshake(RpcCallback &handler) = 0;
virtual void Disconnect() = 0;
void StartReading();
@@ -157,18 +160,33 @@ class RpcConnection : public std::enable_shared_from_this<RpcConnection> {
};
- LockFreeRpcEngine *const engine_;
+ // Initial handshaking protocol: connect->handshake-->(auth)?-->context->connected
+ virtual void SendHandshake(RpcCallback &handler) = 0;
+ void HandshakeComplete(const Status &s);
+ void AuthComplete(const Status &s, const AuthInfo & new_auth_info);
+ void AuthComplete_locked(const Status &s, const AuthInfo & new_auth_info);
+ virtual void SendContext(RpcCallback &handler) = 0;
+ void ContextComplete(const Status &s);
+
+
virtual void OnSendCompleted(const ::asio::error_code &ec,
size_t transferred) = 0;
virtual void OnRecvCompleted(const ::asio::error_code &ec,
size_t transferred) = 0;
virtual void FlushPendingRequests()=0; // Synchronously write the next request
+ void AsyncRpc_locked(
+ const std::string &method_name,
+ const ::google::protobuf::MessageLite *req,
+ std::shared_ptr<::google::protobuf::MessageLite> resp,
+ const RpcCallback &handler);
+ void SendRpcRequests(const std::vector<std::shared_ptr<Request> > & requests);
void AsyncFlushPendingRequests(); // Queue requests to be flushed at a later time
std::shared_ptr<std::string> PrepareHandshakePacket();
+ std::shared_ptr<std::string> PrepareContextPacket();
static std::string SerializeRpcRequest(
const std::string &method_name,
const ::google::protobuf::MessageLite *req);
@@ -180,23 +198,31 @@ class RpcConnection : public std::enable_shared_from_this<RpcConnection> {
void ClearAndDisconnect(const ::asio::error_code &ec);
std::shared_ptr<Request> RemoveFromRunningQueue(int call_id);
- std::shared_ptr<Response> response_;
+ LockFreeRpcEngine *const engine_;
+ std::shared_ptr<Response> current_response_state_;
+ AuthInfo auth_info_;
// Connection can have deferred connection, especially when we're pausing
// during retry
enum ConnectedState {
kNotYetConnected,
kConnecting,
+ kAuthenticating,
kConnected,
kDisconnected
};
static std::string ToString(ConnectedState connected);
-
ConnectedState connected_;
+
+ // State machine for performing a SASL handshake
+ std::shared_ptr<SaslProtocol> sasl_protocol_;
// The request being sent over the wire; will also be in requests_on_fly_
std::shared_ptr<Request> request_over_the_wire_;
// Requests to be sent over the wire
std::vector<std::shared_ptr<Request>> pending_requests_;
+ // Requests to be sent over the wire during authentication; not retried if
+ // there is a connection error
+ std::vector<std::shared_ptr<Request>> auth_requests_;
// Requests that are waiting for responses
typedef std::unordered_map<int, std::shared_ptr<Request>> RequestOnFlyMap;
RequestOnFlyMap requests_on_fly_;
@@ -206,6 +232,8 @@ class RpcConnection : public std::enable_shared_from_this<RpcConnection> {
// Lock for mutable parts of this class that need to be thread safe
std::mutex connection_state_lock_;
+
+ friend class SaslProtocol;
};
@@ -248,7 +276,8 @@ class RpcEngine : public LockFreeRpcEngine {
kCallIdAuthorizationFailed = -1,
kCallIdInvalid = -2,
kCallIdConnectionContext = -3,
- kCallIdPing = -4
+ kCallIdPing = -4,
+ kCallIdSasl = -33
};
RpcEngine(::asio::io_service *io_service, const Options &options,
@@ -286,7 +315,7 @@ class RpcEngine : public LockFreeRpcEngine {
void TEST_SetRpcConnection(std::shared_ptr<RpcConnection> conn);
const std::string &client_name() const override { return client_name_; }
- const std::string &user_name() const override { return user_name_; }
+ const std::string &user_name() const override { return auth_info_.getUser(); }
const std::string &protocol_name() const override { return protocol_name_; }
int protocol_version() const override { return protocol_version_; }
::asio::io_service &io_service() override { return *io_service_; }
@@ -307,10 +336,10 @@ private:
::asio::io_service * const io_service_;
const Options options_;
const std::string client_name_;
- const std::string user_name_;
const std::string protocol_name_;
const int protocol_version_;
const std::unique_ptr<const RetryPolicy> retry_policy_; //null --> no retry
+ AuthInfo auth_info_;
std::string cluster_name_;
std::atomic_int call_id_;
::asio::deadline_timer retry_timer;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_engine.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_engine.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_engine.cc
new file mode 100644
index 0000000..3bb067a
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_engine.cc
@@ -0,0 +1,185 @@
+/**
+ * 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.
+ */
+
+#include "sasl_engine.h"
+
+#include "common/logging.h"
+
+#include <sstream>
+
+
+namespace hdfs {
+
+/*****************************************************************************
+ * BASE CLASS
+ */
+
+SaslEngine::State SaslEngine::getState()
+{
+ return state_;
+}
+
+SaslEngine::~SaslEngine() {
+}
+
+Status SaslEngine::setKerberosInfo(const std::string &principal)
+{
+ principal_ = principal;
+ return Status::OK();
+}
+
+Status SaslEngine::setPasswordInfo(const std::string &id,
+ const std::string &password)
+{
+ id_ = id;
+ password_ = password;
+ return Status::OK();
+}
+
+
+#ifdef USE_GSASL
+/*****************************************************************************
+ * GSASL
+ */
+
+#include <gsasl.h>
+
+
+/*****************************************************************************
+ * UTILITY FUNCTIONS
+ */
+
+Status gsasl_rc_to_status(int rc)
+{
+ if (rc == GSASL_OK) {
+ return Status::OK();
+ } else {
+ std::ostringstream ss;
+ ss << "Cannot initialize client (" << rc << "): " << gsasl_strerror(rc);
+ return Status::Error(ss.str().c_str());
+ }
+}
+
+
+GSaslEngine::~GSaslEngine()
+{
+ if (session_ != nullptr) {
+ gsasl_finish(session_);
+ }
+
+ if (ctx_ != nullptr) {
+ gsasl_done(ctx_);
+ }
+}
+
+std::pair<Status, SaslMethod> GSaslEngine::start(const std::vector<SaslMethod> &protocols)
+{
+ int rc = gsasl_init(&ctx_);
+ if (rc != GSASL_OK) {
+ state_ = kError;
+ return std::make_pair(gsasl_rc_to_status(rc), SaslMethod());
+ }
+
+ // Hack to only do GSSAPI at the moment
+ for (auto protocol: protocols) {
+ if (protocol.mechanism == "GSSAPI") {
+ Status init = init_kerberos(protocol);
+ if (init.ok()) {
+ state_ = kWaitingForData;
+ return std::make_pair(init, protocol);
+ } else {
+ state_ = kError;
+ return std::make_pair(init, SaslMethod());
+ }
+ }
+ }
+
+ state_ = kError;
+ return std::make_pair(Status::Error("No good protocol"), SaslMethod());
+}
+
+Status GSaslEngine::init_kerberos(const SaslMethod & mechanism) {
+ /* Create new authentication session. */
+ int rc = gsasl_client_start(ctx_, mechanism.mechanism.c_str(), &session_);
+ if (rc != GSASL_OK) {
+ return gsasl_rc_to_status(rc);
+ }
+
+ if (!principal_) {
+ return Status::Error("Attempted kerberos authentication with no principal");
+ }
+
+ gsasl_property_set(session_, GSASL_SERVICE, mechanism.protocol.c_str());
+ gsasl_property_set(session_, GSASL_AUTHID, principal_.value().c_str());
+ gsasl_property_set(session_, GSASL_HOSTNAME, mechanism.serverid.c_str());
+ return Status::OK();
+ }
+
+ std::pair<Status, std::string> GSaslEngine::step(const std::string data)
+ {
+ if (state_ != kWaitingForData)
+ LOG_WARN(kRPC, << "GSaslEngine::step when state is " << state_);
+
+ char * output = NULL;
+ size_t outputSize;
+ int rc = gsasl_step(session_, data.c_str(), data.size(), &output,
+ &outputSize);
+
+ if (rc == GSASL_NEEDS_MORE || rc == GSASL_OK) {
+ std::string retval(output, output ? outputSize : 0);
+ if (output) {
+ free(output);
+ }
+
+ if (rc == GSASL_OK) {
+ state_ = kSuccess;
+ }
+
+ return std::make_pair(Status::OK(), retval);
+ }
+ else {
+ if (output) {
+ free(output);
+ }
+ state_ = kFailure;
+ return std::make_pair(gsasl_rc_to_status(rc), "");
+ }
+ }
+
+Status GSaslEngine::finish()
+{
+ if (state_ != kSuccess && state_ != kFailure && state_ != kError )
+ LOG_WARN(kRPC, << "GSaslEngine::finish when state is " << state_);
+
+ if (session_ != nullptr) {
+ gsasl_finish(session_);
+ session_ = NULL;
+ }
+
+ if (ctx_ != nullptr) {
+ gsasl_done(ctx_);
+ ctx_ = nullptr;
+ }
+
+ return Status::OK();
+}
+#endif // USE_GSASL
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_engine.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_engine.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_engine.h
new file mode 100644
index 0000000..e38c30e
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_engine.h
@@ -0,0 +1,125 @@
+/**
+ * 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.
+ */
+
+#ifndef LIB_RPC_SASLENGINE_H
+#define LIB_RPC_SASLENGINE_H
+
+#include "hdfspp/status.h"
+#include "optional.hpp"
+
+#ifdef USE_GSASL
+#include "gsasl.h"
+#endif
+
+#include <vector>
+
+namespace hdfs {
+
+template <class T>
+using optional = std::experimental::optional<T>;
+
+class SaslMethod {
+public:
+ std::string protocol;
+ std::string mechanism;
+ std::string serverid;
+ void * data;
+};
+
+class SaslEngine {
+public:
+ enum State {
+ kUnstarted,
+ kWaitingForData,
+ kSuccess,
+ kFailure,
+ kError,
+ };
+
+ // State transitions:
+ // \--------------------------/
+ // kUnstarted --start--> kWaitingForData --step-+--> kSuccess --finish--v
+ // \-> kFailure -/
+
+ SaslEngine() : state_ (kUnstarted) {}
+ virtual ~SaslEngine();
+
+ // Must be called when state is kUnstarted
+ Status setKerberosInfo(const std::string &principal);
+ // Must be called when state is kUnstarted
+ Status setPasswordInfo(const std::string &id,
+ const std::string &password);
+
+ // Returns the current state
+ State getState();
+
+ // Must be called when state is kUnstarted
+ virtual std::pair<Status,SaslMethod> start(
+ const std::vector<SaslMethod> &protocols) = 0;
+
+ // Must be called when state is kWaitingForData
+ // Returns kOK and any data that should be sent to the server
+ virtual std::pair<Status,std::string> step(const std::string data) = 0;
+
+ // Must only be called when state is kSuccess, kFailure, or kError
+ virtual Status finish() = 0;
+protected:
+ State state_;
+
+ optional<std::string> principal_;
+ optional<std::string> id_;
+ optional<std::string> password_;
+
+};
+
+#ifdef USE_GSASL
+class GSaslEngine : public SaslEngine
+{
+public:
+ GSaslEngine() : SaslEngine(), ctx_(nullptr), session_(nullptr) {}
+ virtual ~GSaslEngine();
+
+ virtual std::pair<Status,SaslMethod> start(
+ const std::vector<SaslMethod> &protocols);
+ virtual std::pair<Status,std::string> step(const std::string data);
+ virtual Status finish();
+private:
+ Gsasl * ctx_;
+ Gsasl_session * session_;
+
+ Status init_kerberos(const SaslMethod & mechanism);
+};
+#endif
+
+#ifdef USE_CYRUS_SASL
+class CyrusSaslEngine : public SaslEngine
+{
+public:
+ GSaslEngine() : SaslEngine(), ctx_(nullptr), session_(nullptr) {}
+ virtual ~GSaslEngine();
+
+ virtual std::pair<Status,SaslMethod> start(
+ const std::vector<SaslMethod> &protocols);
+ virtual std::pair<Status,std::string> step(const std::string data);
+ virtual Status finish();
+private:
+};
+#endif
+
+}
+#endif /* LIB_RPC_SASLENGINE_H */
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_protocol.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_protocol.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_protocol.cc
new file mode 100644
index 0000000..31be77c
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_protocol.cc
@@ -0,0 +1,320 @@
+/**
+ * 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.
+ */
+
+#include "sasl_protocol.h"
+#include "sasl_engine.h"
+#include "rpc_engine.h"
+#include "common/logging.h"
+
+#include <optional.hpp>
+
+namespace hdfs {
+
+using namespace hadoop::common;
+using namespace google::protobuf;
+template <class T>
+using optional = std::experimental::optional<T>;
+
+/*****
+ * Threading model: all entry points need to acquire the sasl_lock before accessing
+ * members of the class
+ *
+ * Lifecycle model: asio may have outstanding callbacks into this class for arbitrary
+ * amounts of time, so any references to the class must be shared_ptr's. The
+ * SASLProtocol keeps a weak_ptr to the owning RpcConnection, which might go away,
+ * so the weak_ptr should be locked only long enough to make callbacks into the
+ * RpcConnection.
+ */
+
+SaslProtocol::SaslProtocol(const std::string & cluster_name,
+ const AuthInfo & auth_info,
+ std::shared_ptr<RpcConnection> connection) :
+ state_(kUnstarted),
+ cluster_name_(cluster_name),
+ auth_info_(auth_info),
+ connection_(connection)
+{
+}
+
+SaslProtocol::~SaslProtocol()
+{
+ std::lock_guard<std::mutex> state_lock(sasl_state_lock_);
+ event_handlers_->call("SASL End", cluster_name_.c_str(), 0);
+}
+
+void SaslProtocol::SetEventHandlers(std::shared_ptr<LibhdfsEvents> event_handlers) {
+ std::lock_guard<std::mutex> state_lock(sasl_state_lock_);
+ event_handlers_ = event_handlers;
+}
+
+void SaslProtocol::authenticate(std::function<void(const Status & status, const AuthInfo new_auth_info)> callback)
+{
+ std::lock_guard<std::mutex> state_lock(sasl_state_lock_);
+
+ LOG_TRACE(kRPC, << "Authenticating as " << auth_info_.getUser());
+
+ assert(state_ == kUnstarted);
+ event_handlers_->call("SASL Start", cluster_name_.c_str(), 0);
+
+ callback_ = callback;
+ state_ = kNegotiate;
+
+ std::shared_ptr<RpcSaslProto> req_msg = std::make_shared<RpcSaslProto>();
+ req_msg->set_state(RpcSaslProto_SaslState_NEGOTIATE);
+
+ // We cheat here since this is always called while holding the RpcConnection's lock
+ std::shared_ptr<RpcConnection> connection = connection_.lock();
+ if (!connection) {
+ return;
+ }
+
+ std::shared_ptr<RpcSaslProto> resp_msg = std::make_shared<RpcSaslProto>();
+ auto self(shared_from_this());
+ connection->AsyncRpc_locked(SASL_METHOD_NAME, req_msg.get(), resp_msg,
+ [self, req_msg, resp_msg] (const Status & status) { self->OnServerResponse(status, resp_msg.get()); } );
+}
+
+AuthInfo::AuthMethod ParseMethod(const std::string & method)
+ {
+ if (0 == strcasecmp(method.c_str(), "SIMPLE")) {
+ return AuthInfo::kSimple;
+ }
+ else if (0 == strcasecmp(method.c_str(), "KERBEROS")) {
+ return AuthInfo::kKerberos;
+ }
+ else if (0 == strcasecmp(method.c_str(), "TOKEN")) {
+ return AuthInfo::kToken;
+ }
+ else {
+ return AuthInfo::kUnknownAuth;
+ }
+}
+
+void SaslProtocol::Negotiate(const hadoop::common::RpcSaslProto * response)
+{
+ std::vector<SaslMethod> protocols;
+
+ bool simple_available = false;
+
+#if defined USE_SASL
+ #if defined USE_CYRUS_SASL
+ sasl_engine_.reset(new CyrusSaslEngine());
+ #elif defined USE_GSASL
+ sasl_engine_.reset(new GSaslEngine());
+ #else
+ #error USE_SASL defined but no engine (USE_GSASL) defined
+ #endif
+#endif
+ if (auth_info_.getToken()) {
+ sasl_engine_->setPasswordInfo(auth_info_.getToken().value().identifier,
+ auth_info_.getToken().value().password);
+ }
+ sasl_engine_->setKerberosInfo(auth_info_.getUser()); // HDFS-10451 will look up principal by username
+
+
+ auto auths = response->auths();
+ for (int i = 0; i < auths.size(); ++i) {
+ auto auth = auths.Get(i);
+ AuthInfo::AuthMethod method = ParseMethod(auth.method());
+
+ switch(method) {
+ case AuthInfo::kToken:
+ case AuthInfo::kKerberos: {
+ SaslMethod new_method;
+ new_method.mechanism = auth.mechanism();
+ new_method.protocol = auth.protocol();
+ new_method.serverid = auth.serverid();
+ new_method.data = const_cast<RpcSaslProto_SaslAuth *>(&response->auths().Get(i));
+ protocols.push_back(new_method);
+ }
+ break;
+ case AuthInfo::kSimple:
+ simple_available = true;
+ break;
+ case AuthInfo::kUnknownAuth:
+ LOG_WARN(kRPC, << "Unknown auth method " << auth.method() << "; ignoring");
+ break;
+ default:
+ LOG_WARN(kRPC, << "Invalid auth type: " << method << "; ignoring");
+ break;
+ }
+ }
+
+ if (!protocols.empty()) {
+ auto init = sasl_engine_->start(protocols);
+ if (init.first.ok()) {
+ auto chosen_auth = reinterpret_cast<RpcSaslProto_SaslAuth *>(init.second.data);
+
+ // Prepare initiate message
+ RpcSaslProto initiate;
+ initiate.set_state(RpcSaslProto_SaslState_INITIATE);
+ RpcSaslProto_SaslAuth * respAuth = initiate.add_auths();
+ respAuth->CopyFrom(*chosen_auth);
+
+ LOG_TRACE(kRPC, << "Using auth: " << chosen_auth->protocol() << "/" <<
+ chosen_auth->mechanism() << "/" << chosen_auth->serverid());
+
+ std::string challenge = chosen_auth->has_challenge() ? chosen_auth->challenge() : "";
+ auto sasl_challenge = sasl_engine_->step(challenge);
+
+ if (sasl_challenge.first.ok()) {
+ if (!sasl_challenge.second.empty()) {
+ initiate.set_token(sasl_challenge.second);
+ }
+
+ std::shared_ptr<RpcSaslProto> return_msg = std::make_shared<RpcSaslProto>();
+ SendSaslMessage(initiate);
+ return;
+ } else {
+ AuthComplete(sasl_challenge.first, auth_info_);
+ return;
+ }
+ } else if (!simple_available) {
+ // If simple IS available, fall through to below
+ AuthComplete(init.first, auth_info_);
+ return;
+ }
+ }
+
+ // There were no protocols, or the SaslEngine couldn't make one work
+ if (simple_available) {
+ // Simple was the only one we could use. That's OK.
+ AuthComplete(Status::OK(), auth_info_);
+ return;
+ } else {
+ // We didn't understand any of the protocols; give back some information
+ std::stringstream ss;
+ ss << "Client cannot authenticate via: ";
+
+ for (int i = 0; i < auths.size(); ++i) {
+ auto auth = auths.Get(i);
+ ss << auth.mechanism() << ", ";
+ }
+
+ AuthComplete(Status::Error(ss.str().c_str()), auth_info_);
+ return;
+ }
+}
+
+void SaslProtocol::Challenge(const hadoop::common::RpcSaslProto * challenge)
+{
+ if (!sasl_engine_) {
+ AuthComplete(Status::Error("Received challenge before negotiate"), auth_info_);
+ return;
+ }
+
+ RpcSaslProto response;
+ response.CopyFrom(*challenge);
+ response.set_state(RpcSaslProto_SaslState_RESPONSE);
+
+ std::string challenge_token = challenge->has_token() ? challenge->token() : "";
+ auto sasl_response = sasl_engine_->step(challenge_token);
+
+ if (sasl_response.first.ok()) {
+ response.set_token(sasl_response.second);
+
+ std::shared_ptr<RpcSaslProto> return_msg = std::make_shared<RpcSaslProto>();
+ SendSaslMessage(response);
+ } else {
+ AuthComplete(sasl_response.first, auth_info_);
+ return;
+ }
+}
+
+bool SaslProtocol::SendSaslMessage(RpcSaslProto & message)
+{
+ assert(lock_held(sasl_state_lock_)); // Must be holding lock before calling
+
+ // RpcConnection might have been freed when we weren't looking. Lock it
+ // to make sure it's there long enough for us
+ std::shared_ptr<RpcConnection> connection = connection_.lock();
+ if (!connection) {
+ LOG_DEBUG(kRPC, << "Tried sending a SASL Message but the RPC connection was gone");
+ return false;
+ }
+
+ std::shared_ptr<RpcSaslProto> resp_msg = std::make_shared<RpcSaslProto>();
+ auto self(shared_from_this());
+ connection->AsyncRpc(SASL_METHOD_NAME, &message, resp_msg,
+ [self, resp_msg] (const Status & status) {
+ self->OnServerResponse(status, resp_msg.get());
+ } );
+
+ return true;
+}
+
+bool SaslProtocol::AuthComplete(const Status & status, const AuthInfo & auth_info)
+{
+ assert(lock_held(sasl_state_lock_)); // Must be holding lock before calling
+
+ // RpcConnection might have been freed when we weren't looking. Lock it
+ // to make sure it's there long enough for us
+ std::shared_ptr<RpcConnection> connection = connection_.lock();
+ if (!connection) {
+ LOG_DEBUG(kRPC, << "Tried sending an AuthComplete but the RPC connection was gone: " << status.ToString());
+ return false;
+ }
+
+ if (!status.ok()) {
+ auth_info_.setMethod(AuthInfo::kAuthFailed);
+ }
+
+ LOG_TRACE(kRPC, << "AuthComplete: " << status.ToString());
+ connection->AuthComplete(status, auth_info);
+
+ return true;
+}
+
+void SaslProtocol::OnServerResponse(const Status & status, const hadoop::common::RpcSaslProto * response)
+{
+ std::lock_guard<std::mutex> state_lock(sasl_state_lock_);
+ LOG_TRACE(kRPC, << "Received SASL response: " << status.ToString());
+
+ if (status.ok()) {
+ switch(response->state()) {
+ case RpcSaslProto_SaslState_NEGOTIATE:
+ Negotiate(response);
+ break;
+ case RpcSaslProto_SaslState_CHALLENGE:
+ Challenge(response);
+ break;
+ case RpcSaslProto_SaslState_SUCCESS:
+ if (sasl_engine_) {
+ sasl_engine_->finish();
+ }
+ AuthComplete(Status::OK(), auth_info_);
+ break;
+
+ case RpcSaslProto_SaslState_INITIATE: // Server side only
+ case RpcSaslProto_SaslState_RESPONSE: // Server side only
+ case RpcSaslProto_SaslState_WRAP:
+ LOG_ERROR(kRPC, << "Invalid client-side SASL state: " << response->state());
+ AuthComplete(Status::Error("Invalid client-side state"), auth_info_);
+ break;
+ default:
+ LOG_ERROR(kRPC, << "Unknown client-side SASL state: " << response->state());
+ AuthComplete(Status::Error("Unknown client-side state"), auth_info_);
+ break;
+ }
+ } else {
+ AuthComplete(status, auth_info_);
+ }
+}
+
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_protocol.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_protocol.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_protocol.h
new file mode 100644
index 0000000..68191b3
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/sasl_protocol.h
@@ -0,0 +1,81 @@
+/**
+ * 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.
+ */
+
+#ifndef LIB_RPC_SASLPROTOCOL_H
+#define LIB_RPC_SASLPROTOCOL_H
+
+#include "hdfspp/status.h"
+#include "common/auth_info.h"
+#include "common/libhdfs_events_impl.h"
+
+#include <RpcHeader.pb.h>
+
+#include <memory>
+#include <mutex>
+#include <functional>
+
+namespace hdfs {
+
+static constexpr const char * SASL_METHOD_NAME = "sasl message";
+
+class RpcConnection;
+class SaslEngine;
+
+class SaslProtocol : public std::enable_shared_from_this<SaslProtocol>
+{
+public:
+ SaslProtocol(const std::string &cluster_name,
+ const AuthInfo & auth_info,
+ std::shared_ptr<RpcConnection> connection);
+ virtual ~SaslProtocol();
+
+ void SetEventHandlers(std::shared_ptr<LibhdfsEvents> event_handlers);
+
+ // Start the async authentication process. Must be called while holding the
+ // connection lock, but all callbacks will occur outside of the connection lock
+ void authenticate(std::function<void(const Status & status, const AuthInfo new_auth_info)> callback);
+ void OnServerResponse(const Status & status, const hadoop::common::RpcSaslProto * response);
+private:
+ enum State {
+ kUnstarted,
+ kNegotiate,
+ kAuthenticate,
+ kComplete
+ };
+
+ // Lock for access to members of the class
+ std::mutex sasl_state_lock_;
+
+ State state_;
+ const std::string cluster_name_;
+ AuthInfo auth_info_;
+ std::weak_ptr<RpcConnection> connection_;
+ std::function<void(const Status & status, const AuthInfo new_auth_info)> callback_;
+ std::unique_ptr<SaslEngine> sasl_engine_;
+ std::shared_ptr<LibhdfsEvents> event_handlers_;
+
+ bool SendSaslMessage(hadoop::common::RpcSaslProto & message);
+ bool AuthComplete(const Status & status, const AuthInfo & auth_info);
+
+ void Negotiate(const hadoop::common::RpcSaslProto * response);
+ void Challenge(const hadoop::common::RpcSaslProto * response);
+};
+
+}
+
+#endif /* LIB_RPC_SASLPROTOCOL_H */
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0ef898e/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/CMakeLists.txt
index 2235c7b..5018afd 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/CMakeLists.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/CMakeLists.txt
@@ -73,15 +73,15 @@ add_memcheck_test(retry_policy retry_policy_test)
include_directories(${CMAKE_CURRENT_BINARY_DIR})
add_executable(rpc_engine_test rpc_engine_test.cc ${PROTO_TEST_SRCS} ${PROTO_TEST_HDRS})
-target_link_libraries(rpc_engine_test test_common rpc proto common ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} gmock_main ${CMAKE_THREAD_LIBS_INIT})
+target_link_libraries(rpc_engine_test test_common rpc proto common ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} ${SASL_LIBRARIES} gmock_main ${CMAKE_THREAD_LIBS_INIT})
add_memcheck_test(rpc_engine rpc_engine_test)
add_executable(bad_datanode_test bad_datanode_test.cc)
-target_link_libraries(bad_datanode_test rpc reader proto fs bindings_c rpc proto common reader connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} gmock_main ${CMAKE_THREAD_LIBS_INIT})
+target_link_libraries(bad_datanode_test rpc reader proto fs bindings_c rpc proto common reader connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} ${SASL_LIBRARIES} gmock_main ${CMAKE_THREAD_LIBS_INIT})
add_memcheck_test(bad_datanode bad_datanode_test)
add_executable(node_exclusion_test node_exclusion_test.cc)
-target_link_libraries(node_exclusion_test fs gmock_main common ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})
+target_link_libraries(node_exclusion_test fs gmock_main common ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} ${SASL_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})
add_memcheck_test(node_exclusion node_exclusion_test)
add_executable(configuration_test configuration_test.cc)
@@ -93,7 +93,7 @@ target_link_libraries(hdfs_configuration_test common gmock_main ${CMAKE_THREAD_L
add_memcheck_test(hdfs_configuration hdfs_configuration_test)
add_executable(hdfspp_errors_test hdfspp_errors.cc)
-target_link_libraries(hdfspp_errors_test common gmock_main bindings_c fs rpc proto common reader connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} gmock_main ${CMAKE_THREAD_LIBS_INIT})
+target_link_libraries(hdfspp_errors_test common gmock_main bindings_c fs rpc proto common reader connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} ${SASL_LIBRARIES} gmock_main ${CMAKE_THREAD_LIBS_INIT})
add_memcheck_test(hdfspp_errors hdfspp_errors_test)
#This test requires a great deal of Hadoop Java infrastructure to run.
@@ -101,15 +101,15 @@ if(HADOOP_BUILD)
add_library(hdfspp_test_shim_static STATIC hdfs_shim.c libhdfs_wrapper.c libhdfspp_wrapper.cc ${LIBHDFSPP_BINDING_C}/hdfs.cc)
build_libhdfs_test(libhdfs_threaded hdfspp_test_shim_static expect.c test_libhdfs_threaded.c ${OS_DIR}/thread.c)
-link_libhdfs_test(libhdfs_threaded hdfspp_test_shim_static fs reader rpc proto common connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} native_mini_dfs ${JAVA_JVM_LIBRARY})
+link_libhdfs_test(libhdfs_threaded hdfspp_test_shim_static fs reader rpc proto common connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} ${SASL_LIBRARIES} native_mini_dfs ${JAVA_JVM_LIBRARY})
add_libhdfs_test(libhdfs_threaded hdfspp_test_shim_static)
endif(HADOOP_BUILD)
add_executable(hdfs_builder_test hdfs_builder_test.cc)
-target_link_libraries(hdfs_builder_test test_common gmock_main bindings_c fs rpc proto common reader connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} gmock_main ${CMAKE_THREAD_LIBS_INIT})
+target_link_libraries(hdfs_builder_test test_common gmock_main bindings_c fs rpc proto common reader connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} ${SASL_LIBRARIES} gmock_main ${CMAKE_THREAD_LIBS_INIT})
add_memcheck_test(hdfs_builder_test hdfs_builder_test)
add_executable(logging_test logging_test.cc)
-target_link_libraries(logging_test common gmock_main bindings_c fs rpc proto common reader connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} gmock_main ${CMAKE_THREAD_LIBS_INIT})
+target_link_libraries(logging_test common gmock_main bindings_c fs rpc proto common reader connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} ${SASL_LIBRARIES} gmock_main ${CMAKE_THREAD_LIBS_INIT})
add_memcheck_test(logging_test logging_test)
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org