diff --git a/src/brpc/policy/baidu_rpc_protocol.cpp b/src/brpc/policy/baidu_rpc_protocol.cpp index 54e611861c..5701cfebdf 100644 --- a/src/brpc/policy/baidu_rpc_protocol.cpp +++ b/src/brpc/policy/baidu_rpc_protocol.cpp @@ -709,9 +709,10 @@ void ProcessRpcRequest(InputMessageBase* msg_base) { const int req_size = static_cast(msg->payload.size()); if (meta.has_attachment_size()) { - if (req_size < meta.attachment_size()) { + if (meta.attachment_size() < 0 || + req_size < meta.attachment_size()) { cntl->SetFailed(EREQUEST, - "attachment_size=%d is larger than request_size=%d", + "Invalid attachment_size=%d (request_size=%d)", meta.attachment_size(), req_size); break; } @@ -1018,9 +1019,10 @@ void ProcessRpcResponse(InputMessageBase* msg_base) { const int res_size = msg->payload.length(); butil::IOBuf* res_buf_ptr = &msg->payload; if (meta.has_attachment_size()) { - if (meta.attachment_size() > res_size) { + if (meta.attachment_size() < 0 || + meta.attachment_size() > res_size) { cntl->SetFailed( - ERESPONSE, "attachment_size=%d is larger than response_size=%d", + ERESPONSE, "Invalid attachment_size=%d (response_size=%d)", meta.attachment_size(), res_size); break; } diff --git a/test/brpc_channel_unittest.cpp b/test/brpc_channel_unittest.cpp index 047d978c12..2f455ffa88 100644 --- a/test/brpc_channel_unittest.cpp +++ b/test/brpc_channel_unittest.cpp @@ -30,6 +30,7 @@ #include "butil/logging.h" #include "butil/files/temp_file.h" #include "butil/fd_guard.h" +#include "butil/raw_pack.h" #include "brpc/socket.h" #include "brpc/acceptor.h" #include "brpc/server.h" @@ -2420,6 +2421,131 @@ TEST(ResponseSocketTest, real_rpc_responses_match_sending_socket) { server.Join(); } +// RpcMeta.attachment_size is a peer-controlled int32. Before it was +// lower-bounded, a negative value passed the "larger than payload" guard and +// `payload_size - attachment_size` overflowed (INT32_MIN) or produced a bogus +// body/attachment split (any other negative value). +TEST(BaiduAttachmentSizeTest, negative_response_attachment_size_is_rejected) { + const int32_t attachment_sizes[] = { INT32_MIN, -1 }; + for (int32_t attachment_size : attachment_sizes) { + SCOPED_TRACE(attachment_size); + brpc::Controller cntl; + test::EchoResponse res; + cntl._response = &res; + ASSERT_EQ(0, bthread_id_lock_and_reset_range( + cntl.call_id(), nullptr, 2)); + ASSERT_EQ(0, bthread_id_unlock(cntl.current_id())); + brpc::SocketId sending_id; + ASSERT_EQ(0, brpc::Socket::Create(brpc::SocketOptions(), &sending_id)); + ASSERT_EQ(0, brpc::Socket::Address( + sending_id, &cntl._current_call.sending_sock)); + brpc::Socket* socket = cntl._current_call.sending_sock.get(); + socket->ReleaseAdditionalReference(); + + brpc::policy::RpcMeta meta; + meta.set_correlation_id(cntl.current_id().value); + meta.mutable_response()->set_error_code(0); + meta.set_attachment_size(attachment_size); + auto* msg = brpc::policy::MostCommonMessage::Get(); + butil::IOBufAsZeroCopyOutputStream meta_stream(&msg->meta); + ASSERT_TRUE(meta.SerializeToZeroCopyStream(&meta_stream)); + test::EchoResponse response; + response.set_message("must not be delivered"); + butil::IOBufAsZeroCopyOutputStream payload_stream(&msg->payload); + ASSERT_TRUE(response.SerializeToZeroCopyStream(&payload_stream)); + socket->ReAddress(&msg->_socket); + socket->PostponeEOF(); + + brpc::policy::ProcessRpcResponse(msg); + EXPECT_EQ(brpc::ERESPONSE, cntl.ErrorCode()) << cntl.ErrorText(); + EXPECT_TRUE(res.message().empty()); + EXPECT_TRUE(cntl.response_attachment().empty()); + } +} + +TEST(BaiduAttachmentSizeTest, negative_request_attachment_size_is_rejected) { + MyEchoService service; + brpc::Server server; + ASSERT_EQ(0, server.AddService(&service, brpc::SERVER_DOESNT_OWN_SERVICE)); + ASSERT_EQ(0, server.Start(0, nullptr)); + + const int32_t attachment_sizes[] = { INT32_MIN, -1 }; + for (int32_t attachment_size : attachment_sizes) { + SCOPED_TRACE(attachment_size); + int fds[2]; + ASSERT_EQ(0, socketpair(AF_UNIX, SOCK_STREAM, 0, fds)); + butil::fd_guard request_fd(fds[0]); + butil::fd_guard peer_fd(fds[1]); + brpc::SocketOptions options; + options.fd = request_fd; + brpc::SocketId socket_id; + ASSERT_EQ(0, brpc::Socket::Create(options, &socket_id)); + request_fd.release(); + brpc::SocketUniquePtr socket; + ASSERT_EQ(0, brpc::Socket::Address(socket_id, &socket)); + socket->ReleaseAdditionalReference(); + + brpc::policy::RpcMeta meta; + meta.set_correlation_id(1); + meta.mutable_request()->set_service_name( + test::EchoService::descriptor()->full_name()); + meta.mutable_request()->set_method_name("Echo"); + meta.set_attachment_size(attachment_size); + auto* msg = brpc::policy::MostCommonMessage::Get(); + butil::IOBufAsZeroCopyOutputStream meta_stream(&msg->meta); + ASSERT_TRUE(meta.SerializeToZeroCopyStream(&meta_stream)); + test::EchoRequest request; + request.set_message("must not be served"); + butil::IOBufAsZeroCopyOutputStream payload_stream(&msg->payload); + ASSERT_TRUE(request.SerializeToZeroCopyStream(&payload_stream)); + msg->_arg = &server; + socket->ReAddress(&msg->_socket); + socket->PostponeEOF(); + + brpc::policy::ProcessRpcRequest(msg); + + // The error response is written back on the arrival socket: + // [PRPC][body_size][meta_size] followed by the serialized RpcMeta. + char header[12]; + butil::IOBuf received; + const int64_t deadline = butil::cpuwide_time_ms() + 5000; + uint32_t body_size = 0; + uint32_t meta_size = 0; + while (received.size() < sizeof(header) || + received.size() < sizeof(header) + body_size) { + const int64_t remaining = deadline - butil::cpuwide_time_ms(); + ASSERT_GT(remaining, 0); + pollfd pfd = {peer_fd, POLLIN, 0}; + ASSERT_EQ(1, poll(&pfd, 1, remaining)); + char buf[1024]; + const ssize_t n = read(peer_fd, buf, sizeof(buf)); + ASSERT_GT(n, 0); + received.append(buf, n); + if (received.size() >= sizeof(header)) { + received.copy_to(header, sizeof(header)); + ASSERT_EQ(0, memcmp(header, "PRPC", 4)); + butil::RawUnpacker(header + 4) + .unpack32(body_size).unpack32(meta_size); + } + } + received.pop_front(sizeof(header)); + ASSERT_LE(meta_size, body_size); + butil::IOBuf response_meta_buf; + received.cutn(&response_meta_buf, meta_size); + brpc::policy::RpcMeta response_meta; + butil::IOBufAsZeroCopyInputStream wrapper(response_meta_buf); + ASSERT_TRUE(response_meta.ParseFromZeroCopyStream(&wrapper)); + EXPECT_EQ(1, response_meta.correlation_id()); + EXPECT_EQ(brpc::EREQUEST, response_meta.response().error_code()) + << response_meta.response().error_text(); + // The request must not have been served: a failed request carries + // no body after the meta. + EXPECT_EQ(body_size, meta_size); + } + server.Stop(0); + server.Join(); +} + TEST_F(ChannelTest, intrusive_ptr_sanity) { MyShared::nctor = 0; MyShared::ndtor = 0;