Harden core RPC marshalling, decode, and message parsing against corrupted data

include/dsn/cpp/blob.h

@@ -37,6 +37,7 @@
 
 # include <dsn/service_api_c.h>
 # include <memory>
+# include <stdexcept>
 # include <vector>
 # include <cstring>
 
@@ -327,8 +328,7 @@ namespace dsn
         }
         else
         {
-            dassert(false, "read beyond the end of buffer");
-            return 0;
+            throw std::out_of_range("read beyond the end of buffer");
         }
     }
 

include/dsn/cpp/clientlet.h

@@ -38,6 +38,7 @@
 # include <dsn/cpp/serialization.h>
 # include <dsn/cpp/task_helper.h>
 # include <dsn/cpp/function_traits.h>
+# include <utility>
 
 namespace dsn
 { 
@@ -300,7 +301,16 @@ namespace dsn
                     typename is_typed_rpc_callback<TCallback>::response_t response{};
                     if (err == ERR_OK)
                     {
-                        ::dsn::unmarshall(resp, response);
+                        decltype(response) parsed{};
+                        auto decode_err = ::dsn::try_unmarshall(resp, parsed);
+                        if (decode_err != ERR_OK)
+                        {
+                            err = decode_err;
+                        }
+                        else
+                        {
+                            response = std::move(parsed);
+                        }
                     }
                     cb_fwd(err, std::move(response));
                 },
@@ -347,7 +357,13 @@ namespace dsn
             {
                 return nullptr;
             }
-            ::dsn::marshall(msg, std::forward<TRequest>(req));
+            auto err = ::dsn::try_marshall(msg, std::forward<TRequest>(req));
+            if (err != ERR_OK)
+            {
+                derror("marshall request failed: %s", err.to_string());
+                dsn_msg_release_ref(msg);
+                return nullptr;
+            }
             return call(server, msg, owner, std::forward<TCallback>(callback), reply_thread_hash);
         }
 
@@ -382,7 +398,13 @@ namespace dsn
             {
                 return rpc_message_helper(nullptr);
             }
-            ::dsn::marshall(msg, std::forward<TRequest>(req));
+            auto err = ::dsn::try_marshall(msg, std::forward<TRequest>(req));
+            if (err != ERR_OK)
+            {
+                derror("marshall request failed: %s", err.to_string());
+                dsn_msg_release_ref(msg);
+                return rpc_message_helper(nullptr);
+            }
             return rpc_message_helper(msg);
         }
 
@@ -431,7 +453,16 @@ namespace dsn
             result.first = task->error();
             if (task->error() == ::dsn::ERR_OK)
             {
-                ::dsn::unmarshall(task->response(), result.second);
+                TResponse parsed{};
+                auto decode_err = ::dsn::try_unmarshall(task->response(), parsed);
+                if (decode_err != ::dsn::ERR_OK)
+                {
+                    result.first = decode_err;
+                }
+                else
+                {
+                    result.second = std::move(parsed);
+                }
             }
             return result;
         }

include/dsn/cpp/rpc_stream.h

@@ -38,6 +38,7 @@
 # include <dsn/service_api_c.h>
 # include <dsn/cpp/blob.h>
 # include <dsn/cpp/auto_codes.h>
+# include <stdexcept>
 
 namespace dsn
 {
@@ -66,14 +67,20 @@ namespace dsn
 
         void set_read_msg(dsn_message_t msg)
         {
-            dassert(msg != nullptr, "rpc_read_stream::set_read_msg got null message");
+            if (msg == nullptr)
+            {
+                throw std::invalid_argument("rpc_read_stream::set_read_msg got null message");
+            }
 
             assign(msg, false);
 
             void* ptr;
             size_t size;
             bool r = dsn_msg_read_next(msg, &ptr, &size);
-            dassert(r, "read msg must have one segment of buffer ready");
+            if (!r)
+            {
+                throw std::out_of_range("read msg must have one segment of buffer ready");
+            }
 
             blob bb((const char*)ptr, 0, (int)size);
             init(bb);
@@ -121,8 +128,10 @@ namespace dsn
         {
             if (!_last_write_next_committed)
             {
-                dassert(dsn_msg_write_commit(native_handle(), (size_t)(total_size() - _last_write_next_total_size)),
-                        "dsn_msg_write_commit failed");
+                if (!dsn_msg_write_commit(native_handle(), (size_t)(total_size() - _last_write_next_total_size)))
+                {
+                    throw std::runtime_error("dsn_msg_write_commit failed");
+                }
                 _last_write_next_committed = true;
             }
         }
@@ -145,8 +154,14 @@ namespace dsn
 
             void* ptr;
             size_t sz;
-            dassert(dsn_msg_write_next(native_handle(), &ptr, &sz, size), "dsn_msg_write_next failed");
-            dassert(sz >= size, "allocated buffer size must be not less than the required size");
+            if (!dsn_msg_write_next(native_handle(), &ptr, &sz, size))
+            {
+                throw std::runtime_error("dsn_msg_write_next failed");
+            }
+            if (sz < size)
+            {
+                throw std::runtime_error("allocated buffer size must be not less than the required size");
+            }
             bb.assign((const char*)ptr, 0, (int)sz);
 
             _last_write_next_total_size = total_size();

include/dsn/cpp/serialization.h

@@ -37,6 +37,7 @@
 
 # include <string>
 # include <sstream>
+# include <stdexcept>
 # include <dsn/cpp/serialization_helper/dsn_types.h>
 # include <dsn/cpp/rpc_stream.h>
 
@@ -82,7 +83,8 @@ namespace dsn
         switch (fmt)
         {
             THRIFT_MARSHALLER
-        default: dassert(false, serialization::no_registered_function_error_notice(value, fmt).c_str());
+        default:
+            throw std::invalid_argument(serialization::no_registered_function_error_notice(value, fmt));
         }
     }
 
@@ -92,7 +94,8 @@ namespace dsn
         switch (fmt)
         {
             THRIFT_UNMARSHALLER
-        default: dassert(false, serialization::no_registered_function_error_notice(value, fmt).c_str());
+        default:
+            throw std::invalid_argument(serialization::no_registered_function_error_notice(value, fmt));
         }
     }
 #else
@@ -120,42 +123,90 @@ namespace dsn
         switch (fmt) \
         { \
             SerializationType##_MARSHALLER \
-            default: dassert(false, serialization::no_registered_function_error_notice(value, fmt).c_str()); \
+            default: \
+                throw std::invalid_argument(serialization::no_registered_function_error_notice(value, fmt)); \
         } \
     } \
     inline void unmarshall(binary_reader& reader, GType &value, dsn_msg_serialize_format fmt) \
     { \
         switch (fmt) \
         { \
             SerializationType##_UNMARSHALLER \
-            default: dassert(false, serialization::no_registered_function_error_notice(value, fmt).c_str()); \
+            default: throw std::invalid_argument(serialization::no_registered_function_error_notice(value, fmt)); \
         } \
     }
 
     template<typename T>
-    inline void marshall(dsn_message_t msg, const T& val)
+    inline void marshall(dsn_message_t msg,
+                         const T& val,
+                         dsn_msg_serialize_format fmt = DSF_INVALID)
     {
-        dassert(msg != nullptr, "marshall got null message");
+        if (msg == nullptr)
+        {
+            throw std::invalid_argument("marshall got null message");
+        }
+
+        if (fmt == DSF_INVALID)
+        {
+            fmt = dsn_msg_get_serialize_format(msg);
+        }
 
         ::dsn::rpc_write_stream writer(msg);
-        marshall(writer, val, dsn_msg_get_serialize_format(msg));
+        marshall(writer, val, fmt);
     }
 
     template<typename T>
-    inline void marshall(dsn_message_t msg, const T& val, dsn_msg_serialize_format fmt)
+    inline error_code try_marshall(dsn_message_t msg,
+                                   const T& val,
+                                   dsn_msg_serialize_format fmt = DSF_INVALID)
     {
-        dassert(msg != nullptr, "marshall got null message");
-
-        ::dsn::rpc_write_stream writer(msg);
-        marshall(writer, val, fmt);
+        try
+        {
+            marshall(msg, val, fmt);
+            return ERR_OK;
+        }
+        catch (const std::invalid_argument&)
+        {
+            return ERR_INVALID_PARAMETERS;
+        }
+        catch (...)
+        {
+            return ERR_INVALID_DATA;
+        }
     }
 
     template<typename T>
-    inline void unmarshall(dsn_message_t msg, /*out*/ T& val)
+    inline void unmarshall(dsn_message_t msg,
+                           /*out*/ T& val,
+                           dsn_msg_serialize_format fmt = DSF_INVALID)
     {
-        dassert(msg != nullptr, "unmarshall got null message");
+        if (msg == nullptr)
+        {
+            throw std::invalid_argument("unmarshall got null message");
+        }
+
+        if (fmt == DSF_INVALID)
+        {
+            fmt = dsn_msg_get_serialize_format(msg);
+        }
 
         ::dsn::rpc_read_stream reader(msg);
-        unmarshall(reader, val, dsn_msg_get_serialize_format(msg));
+        unmarshall(reader, val, fmt);
+    }
+
+    template<typename T>
+    inline error_code try_unmarshall(dsn_message_t msg,
+                                     /*out*/ T& val,
+                                     dsn_msg_serialize_format fmt = DSF_INVALID)
+    {
+        try
+        {
+            unmarshall(msg, val, fmt);
+            return ERR_OK;
+        }
+        catch (...)
+        {
+            return ERR_INVALID_DATA;
+        }
     }
 }

include/dsn/cpp/serialization_helper/thrift_helper.h

@@ -253,8 +253,12 @@ namespace dsn {
         {
             //the protocol is binary protocol
             auto r = iprot->readI64(reinterpret_cast<int64_t&>(_addr.u.value));
-            dassert(_addr.u.v4.type == HOST_TYPE_INVALID || _addr.u.v4.type == HOST_TYPE_IPV4,
+            if (_addr.u.v4.type != HOST_TYPE_INVALID && _addr.u.v4.type != HOST_TYPE_IPV4)
+            {
+                throw ::apache::thrift::protocol::TProtocolException(
+                    ::apache::thrift::protocol::TProtocolException::INVALID_DATA,
                     "only invalid or ipv4 can be deserialized from binary");
+            }
             return r;
         }
         else