This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new be10f511c Fix use-after-free of parsing contexts in
Socket::DebugSocket and ProcessMongoRequest (#3631)
be10f511c is described below
commit be10f511c9fb8d490841cec027b54e706433766d
Author: Weibing Wang <[email protected]>
AuthorDate: Sun Nov 1 00:42:42 2026 +0800
Fix use-after-free of parsing contexts in Socket::DebugSocket and
ProcessMongoRequest (#3631)
* Fix use-after-free of parsing context in Socket::DebugSocket
Socket::DebugSocket() (/sockets/<SocketId>) describes the parsing context
from the bthread serving the request, while the bthread parsing messages
of the socket may reset or release the context at the same time, e.g.
InputMessenger resetting it before trying other protocols. The context
could be destroyed while being described.
Guard reset_parsing_context()/release_parsing_context() and the describing
in DebugSocket() with a mutex, so that a context is taken out of the
socket only after DebugSocket() is done with it.
* Fix use-after-free of mongo context in ProcessMongoRequest
ProcessMongoRequest() runs in a different bthread from the one parsing
messages of the socket, but took the MongoContext from the parsing
context of the socket, which the parsing bthread may reset in the
meantime, e.g. before trying other protocols.
Reference the context in ParseMongoMessage() and carry it with the
request instead.
* Pool mongo input messages
Reuse MostCommonMessage's object pool approach for MongoInputMessage and
mark DestroyImpl as override, so that a mongo request does not need a
heap allocation per request.
* Address review comments on parsing context fix
- Document that the controller keeps its own strong reference of the
mongo context, so it outlives the parsed message.
- Include <unistd.h> explicitly for usleep in the socket test.
- Report errors of the resetting pthread through the argument instead of
gtest assertions, and relax the description assertion so unrelated
formatting changes do not break the test.
* Clarify parsing context ownership and test ioctl result
- Document that set_mongo_session_data() takes a reference of the mongo
context and that a Describe() implementation runs under the parsing
context mutex.
- Check the result of ioctl in the mongo reset test.
---------
Co-authored-by: Mar <[email protected]>
Co-authored-by: brpc-oncall <brpc-oncall@localhost>
---
src/brpc/policy/mongo_protocol.cpp | 42 +++++++++++++---
src/brpc/socket.cpp | 25 +++++-----
src/brpc/socket.h | 7 +++
src/brpc/socket_inl.h | 10 +++-
test/brpc_mongo_protocol_unittest.cpp | 21 ++++++++
test/brpc_socket_unittest.cpp | 91 +++++++++++++++++++++++++++++++++++
6 files changed, 176 insertions(+), 20 deletions(-)
diff --git a/src/brpc/policy/mongo_protocol.cpp
b/src/brpc/policy/mongo_protocol.cpp
index ed8615297..153f1d556 100644
--- a/src/brpc/policy/mongo_protocol.cpp
+++ b/src/brpc/policy/mongo_protocol.cpp
@@ -20,6 +20,7 @@
#include <gflags/gflags.h>
#include "butil/iobuf.h" // butil::IOBuf
+#include "butil/object_pool.h"
#include "butil/strings/string_util.h"
#include "butil/time.h"
@@ -31,7 +32,6 @@
#include "brpc/details/server_private_accessor.h"
#include "brpc/details/controller_private_accessor.h"
#include "brpc/mongo_service_adaptor.h"
-#include "brpc/policy/most_common_message.h"
#include "brpc/policy/nshead_protocol.h"
#include "brpc/policy/mongo.pb.h"
#include "brpc/details/usercode_backup_pool.h"
@@ -110,6 +110,28 @@ void SendMongoResponse::Run() {
}
}
+// A mongo request and the context of its connection. The context is
+// referenced when the request is cut, because the bthread parsing messages
+// may reset the parsing context of the socket before the request is
+// processed in another bthread.
+struct MongoInputMessage : public InputMessageBase {
+ butil::IOBuf meta;
+ butil::IOBuf payload;
+ butil::intrusive_ptr<MongoContext> context;
+
+ inline static MongoInputMessage* Get() {
+ return butil::get_object<MongoInputMessage>();
+ }
+
+ // @InputMessageBase
+ void DestroyImpl() override {
+ meta.clear();
+ payload.clear();
+ context.reset();
+ butil::return_object(this);
+ }
+};
+
ParseResult ParseMongoMessage(butil::IOBuf* source,
Socket* socket, bool /*read_eof*/, const void
*arg) {
const Server* server = static_cast<const Server*>(arg);
@@ -161,13 +183,18 @@ ParseResult ParseMongoMessage(butil::IOBuf* source,
socket_context_msg = new MongoContextMessage(context);
socket->reset_parsing_context(socket_context_msg);
}
- policy::MostCommonMessage* msg = policy::MostCommonMessage::Get();
+ MongoInputMessage* msg = MongoInputMessage::Get();
source->cutn(&msg->meta, sizeof(buf));
size_t act_body_len = source->cutn(&msg->payload, body_len - sizeof(buf));
if (act_body_len != body_len - sizeof(buf)) {
CHECK(false); // Very unlikely, unless memory is corrupted.
return MakeParseError(PARSE_ERROR_TRY_OTHERS);
}
+ MongoContextMessage* context_msg =
+ dynamic_cast<MongoContextMessage*>(socket_context_msg);
+ if (nullptr != context_msg) {
+ msg->context = context_msg->context();
+ }
return MakeMessage(msg);
}
@@ -192,7 +219,7 @@ void EndRunningCallMethodInPool(
::google::protobuf::Closure* done);
void ProcessMongoRequest(InputMessageBase* msg_base) {
- DestroyingPtr<MostCommonMessage>
msg(static_cast<MostCommonMessage*>(msg_base));
+ DestroyingPtr<MongoInputMessage>
msg(static_cast<MongoInputMessage*>(msg_base));
SocketUniquePtr socket_guard(msg->ReleaseSocket());
Socket* socket = socket_guard.get();
const Server* server = static_cast<const Server*>(msg_base->arg());
@@ -212,15 +239,16 @@ void ProcessMongoRequest(InputMessageBase* msg_base) {
const Server::MethodProperty *mp =
server_accessor.FindMethodPropertyByFullName(
srv_des->method(0)->full_name());
- MongoContextMessage* context_msg =
- dynamic_cast<MongoContextMessage*>(socket->parsing_context());
- if (nullptr == context_msg) {
+ if (nullptr == msg->context) {
LOG(WARNING) << "socket context wasn't set correctly";
return;
}
SendMongoResponse* mongo_done = new SendMongoResponse(server);
- mongo_done->cntl.set_mongo_session_data(context_msg->context());
+ // set_mongo_session_data() stores the pointer into a
+ // butil::intrusive_ptr, which takes its own reference. The controller
+ // therefore keeps the context alive after this message is destroyed.
+ mongo_done->cntl.set_mongo_session_data(msg->context.get());
ControllerPrivateAccessor accessor(&(mongo_done->cntl));
const bool security_mode = server->options().security_mode() &&
diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp
index 4c7ad0051..c4411c960 100644
--- a/src/brpc/socket.cpp
+++ b/src/brpc/socket.cpp
@@ -2422,17 +2422,20 @@ void Socket::DebugSocket(std::ostream& os, SocketId id)
{
os << idsizes[i];
}
os << '}';
- Destroyable* const parsing_context = ptr->parsing_context();
- Describable* parsing_context_desc =
dynamic_cast<Describable*>(parsing_context);
- if (parsing_context_desc) {
- os << "\nparsing_context=" << butil::class_name_str(*parsing_context)
<< '{';
- DescribeOptions opt;
- opt.verbose = true;
- IndentingOStream os2(os, 2);
- parsing_context_desc->Describe(os2, opt);
- os << '}';
- } else {
- os << "\nparsing_context=" << ShowObject(parsing_context);
+ {
+ BAIDU_SCOPED_LOCK(ptr->_parsing_context_mutex);
+ Destroyable* const parsing_context = ptr->parsing_context();
+ Describable* parsing_context_desc =
dynamic_cast<Describable*>(parsing_context);
+ if (parsing_context_desc) {
+ os << "\nparsing_context=" <<
butil::class_name_str(*parsing_context) << '{';
+ DescribeOptions opt;
+ opt.verbose = true;
+ IndentingOStream os2(os, 2);
+ parsing_context_desc->Describe(os2, opt);
+ os << '}';
+ } else {
+ os << "\nparsing_context=" << ShowObject(parsing_context);
+ }
}
const SSLState ssl_state = ptr->ssl_state();
os << "\npipeline_q=" << npipelined
diff --git a/src/brpc/socket.h b/src/brpc/socket.h
index 93b92d750..8a1d37229 100644
--- a/src/brpc/socket.h
+++ b/src/brpc/socket.h
@@ -961,6 +961,13 @@ private:
// Saved context for parsing, reset before trying other protocols.
butil::atomic<Destroyable*> _parsing_context;
+ // DebugSocket() describes _parsing_context outside the bthread parsing
+ // messages, which may reset or release the context at the same time.
+ // Both hold this mutex, so that the context is not destroyed while
+ // being described. Describe() runs under the mutex and must not reset
+ // the parsing context of the socket. initialize_parsing_context() does
+ // not need it: it never destroys a context set in the socket.
+ butil::Mutex _parsing_context_mutex;
// The stream whose data the handlers are currently cutting.
butil::atomic<InputMessengerProcessor::StreamType> _parsing_stream_type;
diff --git a/src/brpc/socket_inl.h b/src/brpc/socket_inl.h
index fe6bf76b4..903ef7b62 100644
--- a/src/brpc/socket_inl.h
+++ b/src/brpc/socket_inl.h
@@ -80,14 +80,20 @@ inline void Socket::SetEOF() {
}
inline void Socket::reset_parsing_context(Destroyable* new_context) {
- Destroyable* old_ctx = _parsing_context.exchange(
- new_context, butil::memory_order_acq_rel);
+ Destroyable* old_ctx = nullptr;
+ {
+ // Wait for DebugSocket() which may be describing the old context.
+ BAIDU_SCOPED_LOCK(_parsing_context_mutex);
+ old_ctx = _parsing_context.exchange(
+ new_context, butil::memory_order_acq_rel);
+ }
if (old_ctx) {
old_ctx->Destroy();
}
}
inline Destroyable* Socket::release_parsing_context() {
+ BAIDU_SCOPED_LOCK(_parsing_context_mutex);
return _parsing_context.exchange(nullptr, butil::memory_order_acquire);
}
diff --git a/test/brpc_mongo_protocol_unittest.cpp
b/test/brpc_mongo_protocol_unittest.cpp
index 80bc45435..6ffa064c8 100644
--- a/test/brpc_mongo_protocol_unittest.cpp
+++ b/test/brpc_mongo_protocol_unittest.cpp
@@ -240,6 +240,27 @@ TEST_F(MongoTest, verify_request) {
msg->Destroy();
}
+TEST_F(MongoTest, process_request_after_parsing_context_reset) {
+ brpc::mongo_head_t header = { 0, 0, 0, 0 };
+ header.op_code = brpc::MONGO_OPCODE_REPLY;
+ header.message_length = sizeof(header) + EXP_REQUEST.length();
+ butil::IOBuf total_buf;
+ total_buf.append(static_cast<const void*>(&header), sizeof(header));
+ total_buf.append(EXP_REQUEST);
+ brpc::ParseResult req_pr = brpc::policy::ParseMongoMessage(
+ &total_buf, _socket.get(), false, &_server);
+ ASSERT_EQ(brpc::PARSE_OK, req_pr.error());
+ // The bthread parsing messages may reset the parsing context, e.g. before
+ // trying other protocols, ahead of the request being processed.
+ _socket->reset_parsing_context(nullptr);
+ ProcessMessage(brpc::policy::ProcessMongoRequest, req_pr.message(), false);
+ // The request is still processed with the context it was parsed with.
+ ASSERT_EQ(0ll, _server._nerror_bvar.get_value());
+ int bytes_in_pipe = 0;
+ ASSERT_EQ(0, ioctl(_pipe_fds[0], FIONREAD, &bytes_in_pipe));
+ ASSERT_LT(0, bytes_in_pipe);
+}
+
TEST_F(MongoTest, complete_flow) {
butil::IOBuf request_buf;
butil::IOBuf total_buf;
diff --git a/test/brpc_socket_unittest.cpp b/test/brpc_socket_unittest.cpp
index 48e49a854..0cf581917 100644
--- a/test/brpc_socket_unittest.cpp
+++ b/test/brpc_socket_unittest.cpp
@@ -21,6 +21,7 @@
#include <sys/types.h>
#include <sys/socket.h>
+#include <unistd.h>
#include <fcntl.h> // F_GETFD
#include <gtest/gtest.h>
#include <gflags/gflags.h>
@@ -36,6 +37,7 @@
#include "bthread/unstable.h"
#include "bthread/task_control.h"
#include "brpc/socket.h"
+#include "brpc/describable.h"
#include "brpc/errno.pb.h"
#include "brpc/acceptor.h"
#include "brpc/policy/hulu_pbrpc_protocol.h"
@@ -2418,6 +2420,95 @@ TEST_F(SocketTest, shutdown_write) {
}
}
+// A parsing context taking a while to be described, during which the bthread
+// parsing messages of the socket may reset the context.
+class SlowDescribedContext : public brpc::Destroyable, public
brpc::Describable {
+public:
+ SlowDescribedContext(bthread::CountdownEvent* describing,
+ butil::atomic<int>* ndestroyed)
+ : _describing(describing)
+ , _ndestroyed(ndestroyed)
+ , _name("SlowDescribedContext") {}
+
+ void Describe(std::ostream& os, const brpc::DescribeOptions&) const
override {
+ _describing->signal();
+ // Not bthread_usleep(): DebugSocket() describes under a pthread mutex.
+ usleep(100000);
+ os << _name;
+ }
+
+ void Destroy() override {
+ _ndestroyed->fetch_add(1);
+ delete this;
+ }
+
+private:
+ bthread::CountdownEvent* _describing;
+ butil::atomic<int>* _ndestroyed;
+ std::string _name;
+};
+
+struct DebugSocketArg {
+ brpc::SocketId id;
+ bthread::CountdownEvent* describing;
+ std::string description;
+ int reset_error;
+};
+
+void* DescribeSocket(void* void_arg) {
+ DebugSocketArg* arg = static_cast<DebugSocketArg*>(void_arg);
+ std::ostringstream os;
+ brpc::Socket::DebugSocket(os, arg->id);
+ arg->description = os.str();
+ return nullptr;
+}
+
+void* ResetParsingContextWhileDescribing(void* void_arg) {
+ DebugSocketArg* arg = static_cast<DebugSocketArg*>(void_arg);
+ brpc::SocketUniquePtr s;
+ arg->reset_error = arg->describing->wait();
+ if (arg->reset_error == 0) {
+ arg->reset_error = brpc::Socket::Address(arg->id, &s);
+ }
+ if (arg->reset_error == 0) {
+ // What InputMessenger does before trying other protocols.
+ s->reset_parsing_context(nullptr);
+ }
+ return nullptr;
+}
+
+TEST_F(SocketTest, debug_socket_keeps_parsing_context_alive) {
+ brpc::SocketId id;
+ brpc::SocketOptions options;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ brpc::SocketUniquePtr s;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &s));
+
+ bthread::CountdownEvent describing(1);
+ butil::atomic<int> ndestroyed(0);
+ s->reset_parsing_context(new SlowDescribedContext(&describing,
&ndestroyed));
+
+ DebugSocketArg arg = { id, &describing, "", 0 };
+ bthread_t describer;
+ ASSERT_EQ(0, bthread_start_background(&describer, nullptr, DescribeSocket,
&arg));
+ // Reset in a pthread: signaling a bthread in Describe() would switch to
+ // it on the worker of the describer, which holds a pthread mutex.
+ pthread_t resetter;
+ ASSERT_EQ(0, pthread_create(&resetter, nullptr,
+ ResetParsingContextWhileDescribing, &arg));
+ ASSERT_EQ(0, bthread_join(describer, nullptr));
+ ASSERT_EQ(0, pthread_join(resetter, nullptr));
+
+ ASSERT_EQ(0, arg.reset_error);
+ ASSERT_NE(std::string::npos, arg.description.find("parsing_context="))
+ << arg.description;
+ ASSERT_NE(std::string::npos, arg.description.find("SlowDescribedContext"))
+ << arg.description;
+ ASSERT_EQ(1, ndestroyed.load());
+ ASSERT_EQ(nullptr, s->parsing_context());
+ ASSERT_EQ(0, s->SetFailed());
+}
+
TEST_F(SocketTest, packed_ptr) {
brpc::PackedPtr<int> ptr;
ASSERT_EQ(nullptr, ptr.get());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
