fix(security): fix bug in negotiation_service::on_negotiation_request when rpc_session is closed

src/runtime/security/client_negotiation.cpp

@@ -17,6 +17,7 @@
 
 #include "client_negotiation.h"
 #include "negotiation_utils.h"
+#include "negotiation_service.h"
 
 #include <boost/algorithm/string/join.hpp>
 #include <dsn/dist/fmt_logging.h>
@@ -29,7 +30,7 @@ namespace security {
 DSN_DECLARE_bool(mandatory_auth);
 extern const std::set<std::string> supported_mechanisms;
 
-client_negotiation::client_negotiation(rpc_session *session) : negotiation(session)
+client_negotiation::client_negotiation(rpc_session_ptr session) : negotiation(session)
 {
     _name = fmt::format("CLIENT_NEGOTIATION(SERVER={})", _session->remote_address().to_string());
 }
@@ -179,8 +180,8 @@ void client_negotiation::send(negotiation_status::type status, const blob &msg)
     req->msg = msg;
 
     negotiation_rpc rpc(std::move(req), RPC_NEGOTIATION);
-    rpc.call(_session->remote_address(), nullptr, [this, rpc](error_code err) mutable {
-        handle_response(err, std::move(rpc.response()));
+    rpc.call(_session->remote_address(), nullptr, [rpc](error_code err) mutable {
+        negotiation_service::on_negotiation_response(err, rpc);
     });
 }
 

src/runtime/security/client_negotiation.h

@@ -25,12 +25,12 @@ namespace security {
 class client_negotiation : public negotiation
 {
 public:
-    client_negotiation(rpc_session *session);
+    client_negotiation(rpc_session_ptr session);
 
     void start();
+    void handle_response(error_code err, const negotiation_response &&response);
 
 private:
-    void handle_response(error_code err, const negotiation_response &&response);
     void on_recv_mechanisms(const negotiation_response &resp);
     void on_mechanism_selected(const negotiation_response &resp);
     void on_challenge(const negotiation_response &resp);

src/runtime/security/negotiation.h

@@ -32,7 +32,7 @@ typedef rpc_holder<negotiation_request, negotiation_response> negotiation_rpc;
 class negotiation
 {
 public:
-    negotiation(rpc_session *session)
+    negotiation(rpc_session_ptr session)
         : _session(session), _status(negotiation_status::type::INVALID)
     {
         _sasl = create_sasl_wrapper(_session->is_client());
@@ -49,9 +49,7 @@ class negotiation
     bool check_status(negotiation_status::type status, negotiation_status::type expected_status);
 
 protected:
-    // The ownership of the negotiation instance is held by rpc_session.
-    // So negotiation keeps only a raw pointer.
-    rpc_session *_session;
+    rpc_session_ptr _session;
     std::string _name;
     negotiation_status::type _status;
     std::string _selected_mechanism;

src/runtime/security/negotiation_service.cpp

@@ -18,10 +18,12 @@
 #include "negotiation_service.h"
 #include "negotiation_utils.h"
 #include "server_negotiation.h"
+#include "client_negotiation.h"
 
 #include <dsn/utility/flags.h>
 #include <dsn/tool-api/zlocks.h>
 #include <dsn/dist/failure_detector/fd.code.definition.h>
+#include <dsn/dist/fmt_logging.h>
 
 namespace dsn {
 namespace security {
@@ -52,7 +54,7 @@ void negotiation_service::open_service()
 void negotiation_service::on_negotiation_request(negotiation_rpc rpc)
 {
     dassert(!rpc.dsn_request()->io_session->is_client(),
-            "only server session receive negotiation request");
+            "only server session receives negotiation request");
 
     // reply SASL_AUTH_DISABLE if auth is not enable
     if (!security::FLAGS_enable_auth) {
@@ -67,12 +69,32 @@ void negotiation_service::on_negotiation_request(negotiation_rpc rpc)
             static_cast<server_negotiation *>(_negotiations[rpc.dsn_request()->io_session].get());
     }
 
-    dassert(srv_negotiation != nullptr,
-            "negotiation is null for msg: {}",
-            rpc.dsn_request()->rpc_code().to_string());
+    if (nullptr == srv_negotiation) {
+        derror_f("negotiation is null for msg: {}", rpc.dsn_request()->rpc_code().to_string());
+        return;
+    }
     srv_negotiation->handle_request(rpc);
 }
 
+void negotiation_service::on_negotiation_response(error_code err, negotiation_rpc rpc)
+{
+    dassert(rpc.dsn_request()->io_session->is_client(),
+            "only client session receives negotiation response");
+
+    client_negotiation *cli_negotiation = nullptr;
+    {
+        utils::auto_read_lock l(_lock);
+        cli_negotiation =
+            static_cast<client_negotiation *>(_negotiations[rpc.dsn_request()->io_session].get());
+    }
+
+    if (nullptr == cli_negotiation) {
+        derror_f("negotiation is null for msg: {}", rpc.dsn_request()->rpc_code().to_string());
+        return;
+    }
+    cli_negotiation->handle_response(err, std::move(rpc.response()));
+}
+
 void negotiation_service::on_rpc_connected(rpc_session *session)
 {
     std::unique_ptr<negotiation> nego = security::create_negotiation(session->is_client(), session);

src/runtime/security/negotiation_service.h

@@ -33,6 +33,7 @@ class negotiation_service : public serverlet<negotiation_service>,
     static void on_rpc_disconnected(rpc_session *session);
     static bool on_rpc_recv_msg(message_ex *msg);
     static bool on_rpc_send_msg(message_ex *msg);
+    static void on_negotiation_response(error_code err, negotiation_rpc rpc);
 
     void open_service();
 

src/runtime/security/server_negotiation.cpp

@@ -29,7 +29,7 @@ namespace security {
 DSN_DECLARE_string(service_fqdn);
 DSN_DECLARE_string(service_name);
 
-server_negotiation::server_negotiation(rpc_session *session) : negotiation(session)
+server_negotiation::server_negotiation(rpc_session_ptr session) : negotiation(session)
 {
     _name = fmt::format("SERVER_NEGOTIATION(CLIENT={})", _session->remote_address().to_string());
 }

src/runtime/security/server_negotiation.h

@@ -28,7 +28,7 @@ extern const std::set<std::string> supported_mechanisms;
 class server_negotiation : public negotiation
 {
 public:
-    server_negotiation(rpc_session *session);
+    server_negotiation(rpc_session_ptr session);
 
     void start();
     void handle_request(negotiation_rpc rpc);