fix: improve error handling and validation in various components

Author: noodanee

deps/rocketmq/src/ClientRemotingProcessor.cpp

@@ -63,7 +63,10 @@ RemotingCommand* ClientRemotingProcessor::processRequest(TcpTransportPtr channel
 
 RemotingCommand* ClientRemotingProcessor::checkTransactionState(const std::string& addr, RemotingCommand* request) {
   auto* requestHeader = request->decodeCommandCustomHeader<CheckTransactionStateRequestHeader>();
-  assert(requestHeader != nullptr);
+  if (requestHeader == nullptr) {
+    LOG_ERROR_NEW("Failed to decode CheckTransactionStateRequestHeader");
+    return RemotingCommand::createResponseCommand(ResponseCode::SYSTEM_ERROR, "Invalid request header");
+  }
 
   auto requestBody = request->body();
   if (requestBody != nullptr && requestBody->size() > 0) {

deps/rocketmq/src/message/MessageClientIDSetter.cpp

@@ -70,7 +70,22 @@ void MessageClientIDSetter::setStartTime(uint64_t millis) {
   //   Although not defined, this is almost always an integral value holding the number of seconds
   //   (not counting leap seconds) since 00:00, Jan 1 1970 UTC, corresponding to POSIX time.
   std::time_t tmNow = millis / 1000;
-  std::tm* ptmNow = std::localtime(&tmNow);  // may not be thread-safe
+  std::tm tmResult;
+#ifdef WIN32
+  if (localtime_s(&tmResult, &tmNow) != 0) {
+    start_time_ = millis;
+    next_start_time_ = millis;
+    return;
+  }
+  std::tm* ptmNow = &tmResult;
+#else
+  std::tm* ptmNow = localtime_r(&tmNow, &tmResult);
+  if (ptmNow == nullptr) {
+    start_time_ = millis;
+    next_start_time_ = millis;
+    return;
+  }
+#endif
 
   std::tm curMonthBegin = {0};
   curMonthBegin.tm_year = ptmNow->tm_year;  // since 1900

deps/rocketmq/src/message/MessageDecoder.cpp

@@ -137,7 +137,7 @@ MessageExtPtr MessageDecoder::decode(ByteBuffer& byteBuffer, bool readBody, bool
 
   // 12 STOREHOST
   int storehostIPLength = (sysFlag & MessageSysFlag::STOREHOST_V6_FLAG) == 0 ? kIPv4AddrSize : kIPv6AddrSize;
-  ByteArray storeHost(bornHostLength);
+  ByteArray storeHost(storehostIPLength);
   byteBuffer.get(storeHost, 0, storehostIPLength);
   int32_t storePort = byteBuffer.getInt();
   msgExt->set_store_host(GetSockaddrPtr(IPPortToSockaddr(storeHost, static_cast<uint16_t>(storePort))));
@@ -154,6 +154,15 @@ MessageExtPtr MessageDecoder::decode(ByteBuffer& byteBuffer, bool readBody, bool
   int uncompress_failed = false;
   int32_t bodyLen = byteBuffer.getInt();
   if (bodyLen > 0) {
+    if (bodyLen > byteBuffer.remaining()) {
+      LOG_ERROR_NEW("Invalid bodyLen: {} exceeds buffer size: {}", bodyLen, byteBuffer.remaining());
+      return nullptr;
+    }
+    const int32_t MAX_BODY_SIZE = 4 * 1024 * 1024;
+    if (bodyLen > MAX_BODY_SIZE) {
+      LOG_ERROR_NEW("bodyLen {} exceeds maximum allowed size {}", bodyLen, MAX_BODY_SIZE);
+      return nullptr;
+    }
     if (readBody) {
       ByteArray body(byteBuffer.array() + byteBuffer.arrayOffset() + byteBuffer.position(), bodyLen);
       byteBuffer.position(byteBuffer.position() + bodyLen);

deps/rocketmq/src/transport/EventLoop.cpp

@@ -68,15 +68,15 @@ EventLoop::~EventLoop() {
 }
 
 void EventLoop::start() {
-  if (!is_running_) {
-    is_running_ = true;
+  bool expected = false;
+  if (is_running_.compare_exchange_strong(expected, true)) {
     loop_thread_.start();
   }
 }
 
 void EventLoop::stop() {
-  if (is_running_) {
-    is_running_ = false;
+  bool expected = true;
+  if (is_running_.compare_exchange_strong(expected, false)) {
     loop_thread_.join();
   }
 }

deps/rocketmq/src/transport/EventLoop.h

@@ -21,6 +21,7 @@
 #include <event2/bufferevent.h>
 #include <event2/event.h>
 
+#include <atomic>      // std::atomic
 #include <functional>  // std::function
 #include <memory>      // std::unique_ptr
 
@@ -55,7 +56,7 @@ class EventLoop : public noncopyable {
   struct event_base* event_base_;
   thread loop_thread_;
 
-  bool is_running_;  // aotmic is unnecessary
+  std::atomic<bool> is_running_{false};
 };
 
 class TcpTransport;

deps/rocketmq/src/transport/SocketUtil.cpp

@@ -16,12 +16,14 @@
  */
 #include "SocketUtil.h"
 
-#include <cstdlib>  // std::abort
-#include <cstring>  // std::memcpy, std::memset
+#include <algorithm>  // std::all_of
+#include <cctype>     // std::isdigit
+#include <cstdlib>    // std::abort
+#include <cstring>    // std::memcpy, std::memset
 
 #include <iostream>
 #include <limits>
-#include <memory>    // std::unique_ptr
+#include <memory>     // std::unique_ptr
 #include <stdexcept>  // std::invalid_argument, std::runtime_error
 #include <string>
 
@@ -46,7 +48,7 @@ void SafeMemcpy(void* dest, size_t dest_size, const void* src, size_t src_size)
   if (src_size > dest_size) {
     throw std::invalid_argument("source size exceeds destination buffer size");
   }
-  std::memcpy(dest, src, src_size);
+  std::memmove(dest, src, src_size);
 }
 
 std::unique_ptr<sockaddr_storage> SockaddrToStorage(const sockaddr* src) {
@@ -134,15 +136,18 @@ std::unique_ptr<sockaddr_storage> StringToSockaddr(const std::string& addr) {
   uint16_t port_num = 0;
 
   if (!port_str.empty()) {
+    if (!std::all_of(port_str.begin(), port_str.end(), [](unsigned char c) { return std::isdigit(c); })) {
+      throw std::invalid_argument("port contains non-digit characters: " + port_str);
+    }
     try {
       uint32_t n = std::stoul(port_str);
+      if (n == 0) {
+        throw std::invalid_argument("port cannot be zero");
+      }
       if (n > std::numeric_limits<uint16_t>::max()) {
         throw std::out_of_range("port is too large: " + std::to_string(n) +
                                " (max: " + std::to_string(std::numeric_limits<uint16_t>::max()) + ")");
       }
-      if (n == 0) {
-        throw std::invalid_argument("port cannot be zero");
-      }
       port_num = htons(static_cast<uint16_t>(n));
     } catch (const std::exception& e) {
       throw std::invalid_argument("invalid port: " + port_str + " (" + e.what() + ")");