fix: enhance error handling for empty request body and invalid lengths in message processing

Author: noodanee

deps/rocketmq/src/ClientRemotingProcessor.cpp

@@ -168,6 +168,13 @@ RemotingCommand* ClientRemotingProcessor::receiveReplyMessage(RemotingCommand* r
     }
 
     auto body = request->body();
+    if (body == nullptr) {
+      LOG_WARN_NEW("receive reply message without body");
+      response->set_code(MQResponseCode::SYSTEM_ERROR);
+      response->set_remark("reply message body is empty");
+      return response.release();
+    }
+
     if ((requestHeader->sys_flag() & MessageSysFlag::COMPRESSED_FLAG) == MessageSysFlag::COMPRESSED_FLAG) {
       std::string origin_body;
       if (UtilAll::inflate(*body, origin_body)) {

deps/rocketmq/src/message/MessageDecoder.cpp

@@ -185,15 +185,24 @@ MessageExtPtr MessageDecoder::decode(ByteBuffer& byteBuffer, bool readBody, bool
   }
 
   // 16 TOPIC
-  int8_t topicLen = byteBuffer.get();
+  int8_t topicLenRaw = byteBuffer.get();
+  int topicLen = static_cast<uint8_t>(topicLenRaw);
+  if (topicLen > byteBuffer.remaining()) {
+    LOG_ERROR_NEW("Invalid topic length:{}, remain: {}", topicLen, byteBuffer.remaining());
+    return nullptr;
+  }
   ByteArray topic(topicLen);
   byteBuffer.get(topic);
   msgExt->set_topic(topic.array(), topic.size());
 
   // 17 properties
-  int16_t propertiesLen = byteBuffer.getShort();
-  if (propertiesLen > 0) {
-    ByteArray properties(propertiesLen);
+  int16_t propertiesLenRaw = byteBuffer.getShort();
+  if (propertiesLenRaw < 0 || propertiesLenRaw > byteBuffer.remaining()) {
+    LOG_ERROR_NEW("Invalid properties length:{}, remain:{}", propertiesLenRaw, byteBuffer.remaining());
+    return nullptr;
+  }
+  if (propertiesLenRaw > 0) {
+    ByteArray properties(propertiesLenRaw);
     byteBuffer.get(properties);
     std::string propertiesString(properties.array(), properties.size());
     std::map<std::string, std::string> propertiesMap = string2messageProperties(propertiesString);

deps/rocketmq/src/transport/TcpRemotingClient.cpp

@@ -221,10 +221,11 @@ std::unique_ptr<RemotingCommand> TcpRemotingClient::invokeSync(const std::string
     try {
       doBeforeRpcHooks(addr, request, true);
       auto costTime = UtilAll::currentTimeMillis() - beginStartTime;
-      if (timeoutMillis <= 0 || timeoutMillis < costTime) {
+      if (timeoutMillis <= 0 || timeoutMillis <= costTime) {
         THROW_MQEXCEPTION(RemotingTimeoutException, "invokeSync call timeout", -1);
       }
-      std::unique_ptr<RemotingCommand> response(invokeSyncImpl(channel, request, timeoutMillis));
+      auto remainTimeoutMillis = timeoutMillis - costTime;
+      std::unique_ptr<RemotingCommand> response(invokeSyncImpl(channel, request, remainTimeoutMillis));
       doAfterRpcHooks(addr, request, response.get(), false);
       return response;
     } catch (const RemotingSendRequestException& e) {
@@ -636,7 +637,9 @@ void TcpRemotingClient::processRequestCommand(std::unique_ptr<RemotingCommand> r
 
       doBeforeRpcHooks(channel->getPeerAddrAndPort(), *requestCommand, false);
       response.reset(processor->processRequest(channel, requestCommand.get()));
-      doAfterRpcHooks(channel->getPeerAddrAndPort(), *response, response.get(), true);
+      if (response != nullptr) {
+        doAfterRpcHooks(channel->getPeerAddrAndPort(), *response, response.get(), true);
+      }
     } catch (std::exception& e) {
       LOG_ERROR_NEW("process request exception. {}", e.what());