Merge branch 'master' into recv-mechanism

include/dsn/c/app_model.h

@@ -61,18 +61,18 @@ extern DSN_API bool dsn_mimic_app(const char *app_role, int index);
  start the system with given configuration
 
  \param config           the configuration file for this run
- \param sleep_after_init whether to sleep after rDSN initialization, default is false
+ \param is_server whether it is server or not, default is false
 
  \return true if it succeeds, false if it fails.
  */
-extern DSN_API bool dsn_run_config(const char *config, bool sleep_after_init DEFAULT(false));
+extern DSN_API bool dsn_run_config(const char *config, bool is_server DEFAULT(false));
 
 /*!
  start the system with given arguments
 
  \param argc             argc in C main convention
  \param argv             argv in C main convention
- \param sleep_after_init whether to sleep after rDSN initialization, default is false
+ \param is_server whether it is server or not, default is false
 
  \return true if it succeeds, false if it fails.
 
@@ -89,7 +89,7 @@ extern DSN_API bool dsn_run_config(const char *config, bool sleep_after_init DEF
 
  Note the argc, argv folllows the C main convention that argv[0] is the executable name.
  */
-extern DSN_API void dsn_run(int argc, char **argv, bool sleep_after_init DEFAULT(false));
+extern DSN_API void dsn_run(int argc, char **argv, bool is_server DEFAULT(false));
 
 /*!
  exit the process with the given exit code

include/dsn/tool-api/network.h

@@ -303,6 +303,9 @@ class rpc_session : public ref_counter
 
     void clear_send_queue(bool resend_msgs);
     bool on_disconnected(bool is_write);
+    bool is_auth_success(message_ex *msg);
+    void on_failure(bool is_write = false);
+    void on_success();
 
 protected:
     // constant info

include/dsn/utility/error_code.h

@@ -120,5 +120,6 @@ DEFINE_ERR_CODE(ERR_MOCK_INTERNAL)
 DEFINE_ERR_CODE(ERR_ZOOKEEPER_OPERATION)
 DEFINE_ERR_CODE(ERR_CHILD_REGISTERED)
 DEFINE_ERR_CODE(ERR_INGESTION_FAILED)
-
+DEFINE_ERR_CODE(ERR_UNAUTHENTICATED)
+DEFINE_ERR_CODE(ERR_KRB5_INTERNAL)
 } // namespace dsn

include/dsn/utility/time_utils.h

@@ -61,13 +61,21 @@ inline void time_ms_to_date(uint64_t ts_ms, char *str, int len)
     strftime(str, len, "%Y-%m-%d", get_localtime(ts_ms, &tmp));
 }
 
-// get date string with format of 'yyyy-MM-dd hh:mm:ss' from given timestamp
+// get date string with format of 'yyyy-MM-dd hh:mm:ss' from given timestamp(ms)
 inline void time_ms_to_date_time(uint64_t ts_ms, char *str, int len)
 {
     struct tm tmp;
     strftime(str, len, "%Y-%m-%d %H:%M:%S", get_localtime(ts_ms, &tmp));
 }
 
+// get date string with format of 'yyyy-MM-dd hh:mm:ss' from given timestamp(s)
+inline std::string time_s_to_date_time(uint64_t unix_seconds)
+{
+    char buffer[128];
+    utils::time_ms_to_date_time(unix_seconds * 1000, buffer, 128);
+    return std::string(buffer);
+}
+
 // parse hour/min/sec from the given timestamp
 inline void time_ms_to_date_time(uint64_t ts_ms, int32_t &hour, int32_t &min, int32_t &sec)
 {
@@ -78,12 +86,16 @@ inline void time_ms_to_date_time(uint64_t ts_ms, int32_t &hour, int32_t &min, in
     sec = ret->tm_sec;
 }
 
+// get current physical timestamp in ns
 inline uint64_t get_current_physical_time_ns()
 {
     auto now = std::chrono::high_resolution_clock::now();
     return std::chrono::duration_cast<std::chrono::nanoseconds>(now.time_since_epoch()).count();
 }
 
+// get current physical timestamp in s
+inline uint64_t get_current_physical_time_s() { return get_current_physical_time_ns() * 1e-9; }
+
 // get unix timestamp of today's zero o'clock.
 // eg. `1525881600` returned when called on May 10, 2018, CST
 inline int64_t get_unix_sec_today_midnight()

src/runtime/CMakeLists.txt

@@ -30,5 +30,5 @@ add_library(dsn_runtime STATIC
         tracer.cpp
         zlocks.cpp
         )
-target_link_libraries(dsn_runtime dsn_utils)
+target_link_libraries(dsn_runtime dsn_utils sasl2 gssapi_krb5 krb5)
 install(TARGETS dsn_runtime DESTINATION "lib")

src/runtime/rpc/network.cpp

@@ -25,11 +25,14 @@
  */
 
 #include "runtime/security/negotiation.h"
+#include "runtime/security/negotiation_utils.h"
 #include "message_parser_manager.h"
 #include "runtime/rpc/rpc_engine.h"
 
 #include <dsn/tool-api/network.h>
 #include <dsn/utility/factory_store.h>
+#include <dsn/utility/flags.h>
+#include <dsn/dist/fmt_logging.h>
 
 namespace dsn {
 /*static*/ join_point<void, rpc_session *>
@@ -38,7 +41,7 @@ namespace dsn {
     rpc_session::on_rpc_session_disconnected("rpc.session.disconnected");
 
 namespace security {
-extern bool FLAGS_enable_auth;
+DSN_DECLARE_bool(enable_auth);
 } // namespace security
 
 rpc_session::~rpc_session()
@@ -275,7 +278,11 @@ void rpc_session::send_message(message_ex *msg)
         msg->dl.insert_before(&_messages);
         ++_message_count;
 
-        if (SS_CONNECTED == _connect_state && !_is_sending_next) {
+        // Attention: here we only allow two cases to send message:
+        //  case 1: session's state is SS_CONNECTED
+        //  case 2: session is sending negotiation message
+        if ((SS_CONNECTED == _connect_state || security::is_negotiation_message(msg->rpc_code())) &&
+            !_is_sending_next) {
             _is_sending_next = true;
             sig = _message_sent + 1;
             unlink_message_for_send();
@@ -391,6 +398,19 @@ bool rpc_session::on_disconnected(bool is_write)
     return ret;
 }
 
+bool rpc_session::is_auth_success(message_ex *msg)
+{
+    if (security::FLAGS_enable_auth && !_negotiation->negotiation_succeed()) {
+        dwarn_f("reject message({}) from {}, session {} client",
+                msg->rpc_code().to_string(),
+                _remote_addr.to_string(),
+                is_client() ? "is" : "isn't");
+        return false;
+    }
+
+    return true;
+}
+
 void rpc_session::on_failure(bool is_write)
 {
     if (on_disconnected(is_write)) {
@@ -413,6 +433,16 @@ bool rpc_session::on_recv_message(message_ex *msg, int delay_ms)
     msg->to_address = _net.address();
     msg->io_session = this;
 
+    // return false if msg is negotiation message and auth is not success
+    if (!security::is_negotiation_message(msg->rpc_code()) && !is_auth_success(msg)) {
+        // reply response with ERR_UNAUTHENTICATED if msg is request
+        if (msg->header->context.u.is_request) {
+            _net.engine()->reply(msg->create_response(), ERR_UNAUTHENTICATED);
+        }
+        delete msg;
+        return false;
+    }
+
     if (msg->header->context.u.is_request) {
         // ATTENTION: need to check if self connection occurred.
         //

src/runtime/security/init.cpp

@@ -0,0 +1,64 @@
+// 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 "kinit_context.h"
+
+#include <dsn/dist/fmt_logging.h>
+#include <dsn/utility/flags.h>
+
+namespace dsn {
+namespace security {
+DSN_DECLARE_string(krb5_config);
+DSN_DECLARE_string(krb5_keytab);
+
+/***
+ * set kerberos envs(for more details:
+ * https://web.mit.edu/kerberos/krb5-1.12/doc/admin/env_variables.html)
+ */
+void set_krb5_env(bool is_server)
+{
+    setenv("KRB5CCNAME", is_server ? "MEMORY:pegasus-server" : "MEMORY:pegasus-client", 1);
+    setenv("KRB5_CONFIG", FLAGS_krb5_config, 1);
+    setenv("KRB5_KTNAME", FLAGS_krb5_keytab, 1);
+    setenv("KRB5RCACHETYPE", "none", 1);
+}
+
+error_s init_kerberos(bool is_server)
+{
+    // set kerberos env
+    set_krb5_env(is_server);
+
+    // kinit -k -t <keytab_file> <principal>
+    return run_kinit();
+}
+
+bool init(bool is_server)
+{
+    error_s err = init_kerberos(is_server);
+    if (!err.is_ok()) {
+        derror_f("initialize kerberos failed, with err = {}", err.description());
+        return false;
+    }
+    ddebug("initialize kerberos succeed");
+
+    // TODO(zlw): init sasl
+
+    return true;
+}
+
+} // namespace security
+} // namespace dsn

src/runtime/security/init.h

@@ -0,0 +1,27 @@
+// 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.
+
+#pragma once
+
+#include <dsn/utility/errors.h>
+
+namespace dsn {
+namespace security {
+// init security(kerberos and sasl)
+bool init(bool is_server);
+} // namespace security
+} // namespace dsn