From 25c65e0cde858fb5f5faba87f29079aae79880c8 Mon Sep 17 00:00:00 2001 From: HX Lin <> Date: Thu, 25 Jun 2026 08:17:11 +0800 Subject: [PATCH 1/9] Harden RPC fault injection handling --- include/dsn/cpp/blob.h | 4 +- include/dsn/cpp/clientlet.h | 23 ++++++++- include/dsn/cpp/rpc_stream.h | 11 ++++- include/dsn/cpp/serialization.h | 25 ++++++++-- .../cpp/serialization_helper/thrift_helper.h | 6 ++- include/dsn/cpp/serverlet.h | 49 ++++++++++++++----- src/dev/cpp/utils.cpp | 30 +++++------- src/plugins/tools.common/fault_injector.cpp | 18 ++++++- src/plugins_ext/rDSN.dist.service | 2 +- 9 files changed, 127 insertions(+), 41 deletions(-) diff --git a/include/dsn/cpp/blob.h b/include/dsn/cpp/blob.h index a5e180582..fd277e254 100644 --- a/include/dsn/cpp/blob.h +++ b/include/dsn/cpp/blob.h @@ -37,6 +37,7 @@ # include # include +# include # include # include @@ -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"); } } diff --git a/include/dsn/cpp/clientlet.h b/include/dsn/cpp/clientlet.h index 5d1ae02ac..8d1e0ad63 100644 --- a/include/dsn/cpp/clientlet.h +++ b/include/dsn/cpp/clientlet.h @@ -38,6 +38,7 @@ # include # include # include +# include namespace dsn { @@ -300,7 +301,16 @@ namespace dsn typename is_typed_rpc_callback::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)); }, @@ -431,7 +441,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; } diff --git a/include/dsn/cpp/rpc_stream.h b/include/dsn/cpp/rpc_stream.h index abac7653b..7dfd43377 100644 --- a/include/dsn/cpp/rpc_stream.h +++ b/include/dsn/cpp/rpc_stream.h @@ -38,6 +38,7 @@ # include # include # include +# include 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); diff --git a/include/dsn/cpp/serialization.h b/include/dsn/cpp/serialization.h index 8cedeec86..53257b742 100644 --- a/include/dsn/cpp/serialization.h +++ b/include/dsn/cpp/serialization.h @@ -37,6 +37,7 @@ # include # include +# include # include # include @@ -92,7 +93,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 @@ -128,7 +130,7 @@ namespace dsn 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)); \ } \ } @@ -153,9 +155,26 @@ namespace dsn template inline void unmarshall(dsn_message_t msg, /*out*/ T& val) { - dassert(msg != nullptr, "unmarshall got null message"); + if (msg == nullptr) + { + throw std::invalid_argument("unmarshall got null message"); + } ::dsn::rpc_read_stream reader(msg); unmarshall(reader, val, dsn_msg_get_serialize_format(msg)); } + + template + inline error_code try_unmarshall(dsn_message_t msg, /*out*/ T& val) + { + try + { + unmarshall(msg, val); + return ERR_OK; + } + catch (...) + { + return ERR_INVALID_DATA; + } + } } diff --git a/include/dsn/cpp/serialization_helper/thrift_helper.h b/include/dsn/cpp/serialization_helper/thrift_helper.h index 6d5bb56d5..6374f6cc4 100644 --- a/include/dsn/cpp/serialization_helper/thrift_helper.h +++ b/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(_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 diff --git a/include/dsn/cpp/serverlet.h b/include/dsn/cpp/serverlet.h index 91b0a6918..423ee72e5 100644 --- a/include/dsn/cpp/serverlet.h +++ b/include/dsn/cpp/serverlet.h @@ -123,6 +123,22 @@ namespace dsn TCallback cb; }; + static void reply_invalid_request(dsn_message_t request) + { + auto response = dsn_msg_create_response(request); + if (response == nullptr) + { + derror("dsn_msg_create_response failed"); + return; + } + + auto err = dsn_rpc_reply(response, ERR_INVALID_DATA.get()); + if (err != ERR_OK) + { + derror("dsn_rpc_reply failed: %s", error_code(err).to_string()); + } + } + std::string _name; }; @@ -143,7 +159,7 @@ namespace dsn { if (handler == nullptr) { - dlog(LOG_LEVEL_ERROR, "cpp.serverlet", "register_rpc_handler got null handler"); + derror("register_rpc_handler got null handler"); return false; } @@ -151,7 +167,7 @@ namespace dsn auto hc = (hc_type1*)malloc(sizeof(hc_type1)); if (hc == nullptr) { - dlog(LOG_LEVEL_ERROR, "cpp.serverlet", "failed to allocate rpc handler context"); + derror("failed to allocate rpc handler context"); return false; } hc->this_ = (T*)this; @@ -162,7 +178,10 @@ namespace dsn auto hc2 = (hc_type1*)param; TRequest req; - ::dsn::unmarshall(request, req); + if (::dsn::try_unmarshall(request, req) != ERR_OK) + { + return; + } ((hc2->this_)->*(hc2->cb))(req); }; @@ -179,7 +198,7 @@ namespace dsn { if (handler == nullptr) { - dlog(LOG_LEVEL_ERROR, "cpp.serverlet", "register_rpc_handler got null handler"); + derror("register_rpc_handler got null handler"); return false; } @@ -187,7 +206,7 @@ namespace dsn auto hc = (hc_type2*)malloc(sizeof(hc_type2)); if (hc == nullptr) { - dlog(LOG_LEVEL_ERROR, "cpp.serverlet", "failed to allocate rpc handler context"); + derror("failed to allocate rpc handler context"); return false; } hc->this_ = (T*)this; @@ -198,7 +217,11 @@ namespace dsn auto hc2 = (hc_type2*)param; TRequest req; - ::dsn::unmarshall(request, req); + if (::dsn::try_unmarshall(request, req) != ERR_OK) + { + reply_invalid_request(request); + return; + } TResponse resp; ((hc2->this_)->*(hc2->cb))(req, resp); @@ -220,7 +243,7 @@ namespace dsn { if (handler == nullptr) { - dlog(LOG_LEVEL_ERROR, "cpp.serverlet", "register_async_rpc_handler got null handler"); + derror("register_async_rpc_handler got null handler"); return false; } @@ -228,7 +251,7 @@ namespace dsn auto hc = (hc_type3*)malloc(sizeof(hc_type3)); if (hc == nullptr) { - dlog(LOG_LEVEL_ERROR, "cpp.serverlet", "failed to allocate async rpc handler context"); + derror("failed to allocate async rpc handler context"); return false; } hc->this_ = (T*)this; @@ -239,7 +262,11 @@ namespace dsn auto hc2 = (hc_type3*)param; TRequest req; - ::dsn::unmarshall(request, req); + if (::dsn::try_unmarshall(request, req) != ERR_OK) + { + reply_invalid_request(request); + return; + } rpc_replier replier(dsn_msg_create_response(request)); ((hc2->this_)->*(hc2->cb))(req, replier); @@ -258,7 +285,7 @@ namespace dsn { if (handler == nullptr) { - dlog(LOG_LEVEL_ERROR, "cpp.serverlet", "register_rpc_handler got null handler"); + derror("register_rpc_handler got null handler"); return false; } @@ -266,7 +293,7 @@ namespace dsn auto hc = (hc_type4*)malloc(sizeof(hc_type4)); if (hc == nullptr) { - dlog(LOG_LEVEL_ERROR, "cpp.serverlet", "failed to allocate rpc handler context"); + derror("failed to allocate rpc handler context"); return false; } hc->this_ = (T*)this; diff --git a/src/dev/cpp/utils.cpp b/src/dev/cpp/utils.cpp index b722db8ee..78d04cc6b 100644 --- a/src/dev/cpp/utils.cpp +++ b/src/dev/cpp/utils.cpp @@ -47,6 +47,7 @@ # include # include # include +# include # include # include # include @@ -401,8 +402,7 @@ namespace dsn // now ret should be sizeof(len). if (len < 0) { - derror("binary_reader::read got negative string length: %d", len); - return 0; + throw std::invalid_argument("binary_reader::read got negative string length"); } else if (len == 0) { @@ -417,8 +417,7 @@ namespace dsn } else { - derror("binary_reader::read string beyond the end of buffer"); - return 0; + throw std::out_of_range("binary_reader::read string beyond the end of buffer"); } return ret; @@ -437,8 +436,7 @@ namespace dsn // now ret should be sizeof(len). if (len < 0) { - derror("binary_reader::read got negative blob length: %d", len); - return 0; + throw std::invalid_argument("binary_reader::read got negative blob length"); } else if (len == 0) { @@ -462,8 +460,7 @@ namespace dsn } else { - derror("binary_reader::read blob beyond the end of buffer"); - return 0; + throw std::out_of_range("binary_reader::read blob beyond the end of buffer"); } return ret; @@ -473,8 +470,7 @@ namespace dsn { if (sz < 0) { - dassert(false, "sz is negative: %d", sz); - return 0; + throw std::invalid_argument("binary_reader::read got negative size"); } else if (sz == 0) { @@ -483,7 +479,10 @@ namespace dsn } else if (sz <= get_remaining_size()) { - dassert(buffer != nullptr, "binary_reader::read got null buffer"); + if (buffer == nullptr) + { + throw std::invalid_argument("binary_reader::read got null buffer"); + } memcpy((void*)buffer, _ptr, sz); _ptr += sz; @@ -492,8 +491,7 @@ namespace dsn } else { - derror("binary_reader::read beyond the end of buffer"); - return 0; + throw std::out_of_range("binary_reader::read beyond the end of buffer"); } } @@ -531,8 +529,7 @@ namespace dsn { if (count < 0) { - dassert(false, "count is negative: %d", count); - return false; + throw std::invalid_argument("binary_reader::skip got negative count"); } else if (count <= get_remaining_size()) { @@ -543,8 +540,7 @@ namespace dsn } else { - dassert(false, "read beyond the end of buffer"); - return false; + throw std::out_of_range("read beyond the end of buffer"); } } diff --git a/src/plugins/tools.common/fault_injector.cpp b/src/plugins/tools.common/fault_injector.cpp index faf012189..c5917ad13 100644 --- a/src/plugins/tools.common/fault_injector.cpp +++ b/src/plugins/tools.common/fault_injector.cpp @@ -36,6 +36,7 @@ #include "fault_injector.h" #include +#include # ifdef __TITLE__ # undef __TITLE__ @@ -202,7 +203,13 @@ namespace dsn { static void corrupt_data(message_ex* request, const std::string& corrupt_type) { if (corrupt_type == "header") - replace_value(request->buffers, dsn_random32(0, sizeof(message_header)-1)); + { + static const unsigned int header_mutable_offset = + static_cast(offsetof(message_header, id)); + replace_value(request->buffers, + dsn_random32(header_mutable_offset, + sizeof(message_header) - 1)); + } else if (corrupt_type == "body") { if (request->body_size() == 0) @@ -213,7 +220,14 @@ namespace dsn { replace_value(request->buffers, dsn_random32(0, request->body_size()-1) + sizeof(message_header)); } else if (corrupt_type == "random") - replace_value(request->buffers, dsn_random32(0, request->body_size() + sizeof(message_header) - 1)); + { + if (request->body_size() == 0) + { + dwarn("skip random data corruption for empty message body"); + return; + } + replace_value(request->buffers, dsn_random32(0, request->body_size()-1) + sizeof(message_header)); + } else { derror("try to inject an unknown data corrupt type: %s", corrupt_type.c_str()); diff --git a/src/plugins_ext/rDSN.dist.service b/src/plugins_ext/rDSN.dist.service index ed4bfa2a4..306eac298 160000 --- a/src/plugins_ext/rDSN.dist.service +++ b/src/plugins_ext/rDSN.dist.service @@ -1 +1 @@ -Subproject commit ed4bfa2a4210a1252fb2b7db9700ae0e30ab6d4b +Subproject commit 306eac29882362ddba301ca6d661be1ab7f3487c From 450b1f72e3ea37621a74784c626b150839e3f88b Mon Sep 17 00:00:00 2001 From: HX Lin <> Date: Thu, 25 Jun 2026 08:44:28 +0800 Subject: [PATCH 2/9] Harden RPC marshalling errors --- include/dsn/cpp/clientlet.h | 16 +++++++-- include/dsn/cpp/rpc_stream.h | 16 ++++++--- include/dsn/cpp/serialization.h | 60 +++++++++++++++++++++++++-------- src/dev/cpp/utils.cpp | 49 ++++++++++++++++++++------- 4 files changed, 108 insertions(+), 33 deletions(-) diff --git a/include/dsn/cpp/clientlet.h b/include/dsn/cpp/clientlet.h index 8d1e0ad63..c35c2ac73 100644 --- a/include/dsn/cpp/clientlet.h +++ b/include/dsn/cpp/clientlet.h @@ -357,7 +357,13 @@ namespace dsn { return nullptr; } - ::dsn::marshall(msg, std::forward(req)); + auto err = ::dsn::try_marshall(msg, std::forward(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(callback), reply_thread_hash); } @@ -392,7 +398,13 @@ namespace dsn { return rpc_message_helper(nullptr); } - ::dsn::marshall(msg, std::forward(req)); + auto err = ::dsn::try_marshall(msg, std::forward(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); } diff --git a/include/dsn/cpp/rpc_stream.h b/include/dsn/cpp/rpc_stream.h index 7dfd43377..7e91e5946 100644 --- a/include/dsn/cpp/rpc_stream.h +++ b/include/dsn/cpp/rpc_stream.h @@ -128,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; } } @@ -152,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(); diff --git a/include/dsn/cpp/serialization.h b/include/dsn/cpp/serialization.h index 53257b742..005b818fa 100644 --- a/include/dsn/cpp/serialization.h +++ b/include/dsn/cpp/serialization.h @@ -83,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)); } } @@ -122,7 +123,8 @@ 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) \ @@ -135,41 +137,71 @@ namespace dsn } template - 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 - 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 - 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) { 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 - inline error_code try_unmarshall(dsn_message_t msg, /*out*/ T& val) + inline error_code try_unmarshall(dsn_message_t msg, + /*out*/ T& val, + dsn_msg_serialize_format fmt = DSF_INVALID) { try { - unmarshall(msg, val); + unmarshall(msg, val, fmt); return ERR_OK; } catch (...) diff --git a/src/dev/cpp/utils.cpp b/src/dev/cpp/utils.cpp index 78d04cc6b..ff2907407 100644 --- a/src/dev/cpp/utils.cpp +++ b/src/dev/cpp/utils.cpp @@ -548,7 +548,10 @@ namespace dsn binary_writer::binary_writer(int reserveBufferSize) { - dassert(reserveBufferSize >= 0, "binary_writer got negative reserve buffer size"); + if (reserveBufferSize < 0) + { + throw std::invalid_argument("binary_writer got negative reserve buffer size"); + } _total_size = 0; _buffers.reserve(1); _reserved_size_per_buffer = @@ -663,7 +666,10 @@ namespace dsn void binary_writer::write_empty(int sz) { - dassert(sz >= 0, "binary_writer::write_empty got negative size"); + if (sz < 0) + { + throw std::invalid_argument("binary_writer::write_empty got negative size"); + } if (sz == 0) { return; @@ -685,8 +691,10 @@ namespace dsn allocSize = sz; create_buffer(allocSize); - dassert(_current_buffer != nullptr && _current_buffer_length > 0, - "binary_writer::write_empty failed to create buffer"); + if (_current_buffer == nullptr || _current_buffer_length <= 0) + { + throw std::runtime_error("binary_writer::write_empty failed to create buffer"); + } _current_offset += sz; } @@ -695,12 +703,18 @@ namespace dsn void binary_writer::write(const char* buffer, int sz) { - dassert(sz >= 0, "binary_writer::write got negative size"); + if (sz < 0) + { + throw std::invalid_argument("binary_writer::write got negative size"); + } if (sz == 0) { return; } - dassert(buffer != nullptr, "binary_writer::write got null buffer"); + if (buffer == nullptr) + { + throw std::invalid_argument("binary_writer::write got null buffer"); + } int rem_size = _current_buffer_length - _current_offset; if (rem_size >= sz) @@ -724,8 +738,10 @@ namespace dsn allocSize = sz; create_buffer(allocSize); - dassert(_current_buffer != nullptr && _current_buffer_length > 0, - "binary_writer::write failed to create buffer"); + if (_current_buffer == nullptr || _current_buffer_length <= 0) + { + throw std::runtime_error("binary_writer::write failed to create buffer"); + } memcpy((void*)(_current_buffer + _current_offset), buffer + rem_size, (size_t)sz); _current_offset += sz; _total_size += sz; @@ -734,15 +750,20 @@ namespace dsn bool binary_writer::next(void** data, int* size) { - dassert(data != nullptr && size != nullptr, "binary_writer::next got null output parameter"); + if (data == nullptr || size == nullptr) + { + throw std::invalid_argument("binary_writer::next got null output parameter"); + } int rem_size = _current_buffer_length - _current_offset; if (rem_size == 0) { create_buffer(_reserved_size_per_buffer); rem_size = _current_buffer_length; - dassert(_current_buffer != nullptr && rem_size > 0, - "binary_writer::next failed to create buffer"); + if (_current_buffer == nullptr || rem_size <= 0) + { + throw std::runtime_error("binary_writer::next failed to create buffer"); + } } *size = rem_size; @@ -754,8 +775,10 @@ namespace dsn bool binary_writer::backup(int count) { - dassert(count >= 0 && count <= _current_offset, - "currently we don't support backup before the last buffer's header"); + if (count < 0 || count > _current_offset) + { + throw std::invalid_argument("currently we don't support backup before the last buffer's header"); + } _current_offset -= count; _total_size -= count; return true; From ac41bf8c009823f39042caa8dda189e953174161 Mon Sep 17 00:00:00 2001 From: HX Lin <> Date: Thu, 25 Jun 2026 09:31:12 +0800 Subject: [PATCH 3/9] Harden RPC corruption fault injection --- src/plugins/tools.common/fault_injector.cpp | 7 +++++-- src/plugins_ext/rDSN.dist.service | 2 +- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/plugins/tools.common/fault_injector.cpp b/src/plugins/tools.common/fault_injector.cpp index c5917ad13..0a2db4d17 100644 --- a/src/plugins/tools.common/fault_injector.cpp +++ b/src/plugins/tools.common/fault_injector.cpp @@ -206,9 +206,12 @@ namespace dsn { { static const unsigned int header_mutable_offset = static_cast(offsetof(message_header, id)); + static const unsigned int header_mutable_size = + static_cast(offsetof(message_header, context)) - + header_mutable_offset; replace_value(request->buffers, - dsn_random32(header_mutable_offset, - sizeof(message_header) - 1)); + header_mutable_offset + + dsn_random32(0, header_mutable_size - 1)); } else if (corrupt_type == "body") { diff --git a/src/plugins_ext/rDSN.dist.service b/src/plugins_ext/rDSN.dist.service index 306eac298..e2c4e0627 160000 --- a/src/plugins_ext/rDSN.dist.service +++ b/src/plugins_ext/rDSN.dist.service @@ -1 +1 @@ -Subproject commit 306eac29882362ddba301ca6d661be1ab7f3487c +Subproject commit e2c4e06277283eea16fa0960f6cc07119bd6d38c From 94c0a6d9cbbc4e1dea94a82799bdf80a326b1837 Mon Sep 17 00:00:00 2001 From: HX Lin <> Date: Thu, 25 Jun 2026 10:43:15 +0800 Subject: [PATCH 4/9] Harden core RPC reply decoding against corrupted data rpc_client_matcher::on_recv_reply decodes the ERR_FORWARD_TO_OTHERS forward address via try_unmarshall and delivers ERR_NETWORK_FAILURE to the waiting caller on a decode failure, instead of forwarding to a garbage address. command_manager::run_command returns false when a remote CLI response cannot be decoded, and partition_resolver_simple surfaces a config-response decode failure through client_err rather than throwing past the resolve callback. Update rDSN.dist.service submodule pointer to pick up the meta/replica RPC request-handler hardening. --- src/core/src/command_manager.cpp | 6 +++++- src/core/src/rpc_engine.cpp | 12 ++++++++++-- .../partition_resolver_simple.cpp | 15 +++++++++++++-- src/plugins_ext/rDSN.dist.service | 2 +- 4 files changed, 29 insertions(+), 6 deletions(-) diff --git a/src/core/src/command_manager.cpp b/src/core/src/command_manager.cpp index 7cbef5b97..0ab8b4899 100644 --- a/src/core/src/command_manager.cpp +++ b/src/core/src/command_manager.cpp @@ -375,7 +375,11 @@ namespace dsn { if (resp != nullptr) { std::string o2 = output.c_str(); - ::dsn::unmarshall(resp, o2); + if (::dsn::try_unmarshall(resp, o2) != ERR_OK) + { + dwarn("cli run for %s failed: invalid response", cmd.c_str()); + return false; + } return true; } else diff --git a/src/core/src/rpc_engine.cpp b/src/core/src/rpc_engine.cpp index 2abfc1f54..1ea6908da 100644 --- a/src/core/src/rpc_engine.cpp +++ b/src/core/src/rpc_engine.cpp @@ -159,8 +159,15 @@ namespace dsn { if (err == ERR_FORWARD_TO_OTHERS) { rpc_address addr; - ::dsn::unmarshall((dsn_message_t)reply, addr); - + if (::dsn::try_unmarshall((dsn_message_t)reply, addr) != ERR_OK) + { + derror("invalid forward reply: failed to decode forward address, trace_id = %016" PRIx64, + reply->header->trace_id); + call->set_delay(delay_ms); + call->enqueue(ERR_NETWORK_FAILURE, reply); + } + else + { // handle the case of forwarding to itself where addr == req->to_address. dbg_dassert(addr != req->to_address, "impossible forwarding to myself as this only happens when i'm pure client so i don't get a named to_address %s", @@ -195,6 +202,7 @@ namespace dsn { dassert(reply->get_count() == 0, "reply should not be referenced by anybody so far"); delete reply; + } } else { diff --git a/src/plugins/dist.uri.resolver/partition_resolver_simple.cpp b/src/plugins/dist.uri.resolver/partition_resolver_simple.cpp index fbc35e296..631b14d83 100644 --- a/src/plugins/dist.uri.resolver/partition_resolver_simple.cpp +++ b/src/plugins/dist.uri.resolver/partition_resolver_simple.cpp @@ -296,8 +296,19 @@ namespace dsn if (err == ERR_OK) { configuration_query_by_index_response resp; - unmarshall(response, resp); - if (resp.err == ERR_OK) + auto decode_err = ::dsn::try_unmarshall(response, resp); + if (decode_err != ERR_OK) + { + derror("%s.client: query config reply, gpid = %d.%d, invalid response: %s", + _app_path.c_str(), + _app_id, + partition_index, + decode_err.to_string() + ); + + client_err = decode_err; + } + else if (resp.err == ERR_OK) { zauto_write_lock l(_config_lock); diff --git a/src/plugins_ext/rDSN.dist.service b/src/plugins_ext/rDSN.dist.service index e2c4e0627..c94e3f61d 160000 --- a/src/plugins_ext/rDSN.dist.service +++ b/src/plugins_ext/rDSN.dist.service @@ -1 +1 @@ -Subproject commit e2c4e06277283eea16fa0960f6cc07119bd6d38c +Subproject commit c94e3f61ddc54b6d4ab8800d658e36270f6439ac From 946952c4cf067145ddb01c15565d895277b7e602 Mon Sep 17 00:00:00 2001 From: HX Lin <> Date: Fri, 26 Jun 2026 08:35:41 +0800 Subject: [PATCH 5/9] Add tests for checked serialization helpers Cover try_marshall and try_unmarshall success and failure paths with focused rpc_message tests. The tests verify null-message handling, invalid format handling, and a successful string round-trip through a receive-mode message copy. Update rDSN.dist.service submodule to allow DSF_INVALID for opaque mutation-log records while still rejecting out-of-range serialization values. --- src/core/src/rpc_message.test.cpp | 60 +++++++++++++++++++++++++++++++ src/plugins_ext/rDSN.dist.service | 2 +- 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/src/core/src/rpc_message.test.cpp b/src/core/src/rpc_message.test.cpp index e200eb8d3..486d35d8c 100644 --- a/src/core/src/rpc_message.test.cpp +++ b/src/core/src/rpc_message.test.cpp @@ -343,3 +343,63 @@ TEST(core, dsn_msg_invalid_parameters) dsn_msg_release_ref(request); } + +TEST(core, try_marshall) +{ + const std::string value("hello, rdsn"); + + ASSERT_EQ(ERR_INVALID_PARAMETERS, ::dsn::try_marshall(nullptr, value)); + + dsn_message_t invalid_request = dsn_msg_create_request(RPC_CODE_FOR_TEST, 100, 1, 2); + ASSERT_NE(nullptr, invalid_request); + dsn_msg_add_ref(invalid_request); + + ASSERT_EQ(ERR_INVALID_PARAMETERS, + ::dsn::try_marshall( + invalid_request, value, static_cast(DSF_JSON + 1))); + dsn_msg_release_ref(invalid_request); + + dsn_message_t request = dsn_msg_create_request(RPC_CODE_FOR_TEST, 100, 1, 2); + ASSERT_NE(nullptr, request); + dsn_msg_add_ref(request); + + ASSERT_EQ(ERR_OK, ::dsn::try_marshall(request, value, DSF_THRIFT_BINARY)); + + dsn_message_t received_request = dsn_msg_copy(request, true, true); + ASSERT_NE(nullptr, received_request); + dsn_msg_add_ref(received_request); + + std::string decoded; + ::dsn::unmarshall(received_request, decoded, DSF_THRIFT_BINARY); + ASSERT_EQ(value, decoded); + + dsn_msg_release_ref(received_request); + dsn_msg_release_ref(request); +} + +TEST(core, try_unmarshall) +{ + std::string decoded; + ASSERT_EQ(ERR_INVALID_DATA, ::dsn::try_unmarshall(nullptr, decoded)); + + const std::string value("hello, rdsn"); + dsn_message_t request = dsn_msg_create_request(RPC_CODE_FOR_TEST, 100, 1, 2); + ASSERT_NE(nullptr, request); + dsn_msg_add_ref(request); + + ::dsn::marshall(request, value, DSF_THRIFT_BINARY); + dsn_message_t received_request = dsn_msg_copy(request, true, true); + ASSERT_NE(nullptr, received_request); + dsn_msg_add_ref(received_request); + + ASSERT_EQ(ERR_INVALID_DATA, + ::dsn::try_unmarshall( + received_request, decoded, static_cast(DSF_JSON + 1))); + + decoded.clear(); + ASSERT_EQ(ERR_OK, ::dsn::try_unmarshall(received_request, decoded, DSF_THRIFT_BINARY)); + ASSERT_EQ(value, decoded); + + dsn_msg_release_ref(received_request); + dsn_msg_release_ref(request); +} diff --git a/src/plugins_ext/rDSN.dist.service b/src/plugins_ext/rDSN.dist.service index c94e3f61d..5441af9fc 160000 --- a/src/plugins_ext/rDSN.dist.service +++ b/src/plugins_ext/rDSN.dist.service @@ -1 +1 @@ -Subproject commit c94e3f61ddc54b6d4ab8800d658e36270f6439ac +Subproject commit 5441af9fcd7722764092c6b96f740ad14241ab56 From 986501fb1741b95d2cde3b5a69653f67eb737530 Mon Sep 17 00:00:00 2001 From: HX Lin <> Date: Fri, 26 Jun 2026 09:43:47 +0800 Subject: [PATCH 6/9] Update dist service robustness fixes --- src/plugins_ext/rDSN.dist.service | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/plugins_ext/rDSN.dist.service b/src/plugins_ext/rDSN.dist.service index 5441af9fc..f9827db9f 160000 --- a/src/plugins_ext/rDSN.dist.service +++ b/src/plugins_ext/rDSN.dist.service @@ -1 +1 @@ -Subproject commit 5441af9fcd7722764092c6b96f740ad14241ab56 +Subproject commit f9827db9f3d6c50f0ee6015e7b781606982b4d5c From 930a3d38b891711cb919f6002f4fc22f682d24c2 Mon Sep 17 00:00:00 2001 From: HX Lin <> Date: Fri, 26 Jun 2026 15:15:36 +0800 Subject: [PATCH 7/9] Update dist service config validation --- src/plugins_ext/rDSN.dist.service | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/plugins_ext/rDSN.dist.service b/src/plugins_ext/rDSN.dist.service index f9827db9f..5091dd597 160000 --- a/src/plugins_ext/rDSN.dist.service +++ b/src/plugins_ext/rDSN.dist.service @@ -1 +1 @@ -Subproject commit f9827db9f3d6c50f0ee6015e7b781606982b4d5c +Subproject commit 5091dd597ebe79a1c8a6f72b1f3cc19cae2984a4 From 9319f81416cfd66f6464c64d460d1ea63ab48f37 Mon Sep 17 00:00:00 2001 From: HX Lin <> Date: Fri, 26 Jun 2026 16:05:54 +0800 Subject: [PATCH 8/9] Harden thrift message parsing --- .../tools.common/thrift_message_parser.cpp | 62 ++++++++++++++++--- 1 file changed, 53 insertions(+), 9 deletions(-) diff --git a/src/plugins/tools.common/thrift_message_parser.cpp b/src/plugins/tools.common/thrift_message_parser.cpp index a63676223..aad887972 100644 --- a/src/plugins/tools.common/thrift_message_parser.cpp +++ b/src/plugins/tools.common/thrift_message_parser.cpp @@ -98,6 +98,17 @@ namespace dsn dsn::blob msg_bb = buf.range(0, msg_sz); message_ex* msg = parse_message(_thrift_header, msg_bb); + // parse_message returns null when the (untrusted) thrift body is malformed + // (bad message-begin envelope, over-long rpc name, or a non-request message). + // The receive loop only treats read_next == -1 as a hard failure, so we must + // signal it here instead of falling through and dereferencing the null msg. + if (msg == nullptr) + { + derror("thrift message body check failed"); + read_next = -1; + return nullptr; + } + reader->_buffer = buf.range(msg_sz); reader->_buffer_occupied -= msg_sz; _header_parsed = false; @@ -278,15 +289,38 @@ namespace dsn dsn::message_ex* msg = message_ex::create_receive_message_with_standalone_header(body_data); dsn::message_header* dsn_hdr = msg->header; - dsn::rpc_read_stream stream(msg); - ::dsn::binary_reader_transport binary_transport(stream); - boost::shared_ptr< ::dsn::binary_reader_transport > trans_ptr(&binary_transport, [](::dsn::binary_reader_transport*) {}); - ::apache::thrift::protocol::TBinaryProtocol iprot(trans_ptr); - std::string fname; - ::apache::thrift::protocol::TMessageType mtype; - int32_t seqid; - iprot.readMessageBegin(fname, mtype, seqid); + ::apache::thrift::protocol::TMessageType mtype = ::apache::thrift::protocol::T_CALL; + int32_t seqid = 0; + bool parsed = false; + + // The thrift message-begin envelope is decoded from untrusted network bytes. A corrupt + // length/string can make readMessageBegin throw (TProtocolException / out_of_range from + // the underlying binary_reader). Keep the read stream in an inner scope so it (and its + // dsn_msg_read_commit) is destroyed before we may free msg on the failure path. + { + dsn::rpc_read_stream stream(msg); + ::dsn::binary_reader_transport binary_transport(stream); + boost::shared_ptr< ::dsn::binary_reader_transport > trans_ptr(&binary_transport, [](::dsn::binary_reader_transport*) {}); + ::apache::thrift::protocol::TBinaryProtocol iprot(trans_ptr); + + try + { + iprot.readMessageBegin(fname, mtype, seqid); + parsed = true; + } + catch (std::exception& ex) + { + derror("thrift message begin parse failed: %s", ex.what()); + } + } + + if (!parsed) + { + delete msg; + return nullptr; + } + dinfo("rpc name: %s, type: %d, seqid: %d", fname.c_str(), mtype, seqid); dsn_hdr->hdr_type = THRIFT_HDR_SIG; @@ -299,6 +333,7 @@ namespace dsn if (name_len < 0 || static_cast(name_len) >= sizeof(dsn_hdr->rpc_name)) { derror("thrift rpc name is too long: %s", fname.c_str()); + delete msg; return nullptr; } dsn_hdr->gpid.u.app_id = thrift_header.app_id; @@ -309,7 +344,16 @@ namespace dsn if (mtype == ::apache::thrift::protocol::T_CALL || mtype == ::apache::thrift::protocol::T_ONEWAY) dsn_hdr->context.u.is_request = 1; - dassert(dsn_hdr->context.u.is_request == 1, "only support receive request"); + + // Only requests are accepted here; a corrupt/unexpected message type used to abort via + // dassert. Reject it through the parser's error channel (caller closes the connection) + // instead of crashing the whole process on malformed network input. + if (dsn_hdr->context.u.is_request != 1) + { + derror("thrift message is not a request, type = %d", mtype); + delete msg; + return nullptr; + } dsn_hdr->context.u.serialize_format = DSF_THRIFT_BINARY; // always serialize in thrift binary return msg; From b1634ef131813af3ffef6a34815203d13bfee8a4 Mon Sep 17 00:00:00 2001 From: HX Lin <> Date: Sat, 27 Jun 2026 18:03:25 +0800 Subject: [PATCH 9/9] Update rDSN.dist.service submodule to merged master --- src/plugins_ext/rDSN.dist.service | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/plugins_ext/rDSN.dist.service b/src/plugins_ext/rDSN.dist.service index 5091dd597..b15e9c04d 160000 --- a/src/plugins_ext/rDSN.dist.service +++ b/src/plugins_ext/rDSN.dist.service @@ -1 +1 @@ -Subproject commit 5091dd597ebe79a1c8a6f72b1f3cc19cae2984a4 +Subproject commit b15e9c04d29d84e3ecd9917189056f3327da4290