fix: update Reader::tryToRecover function to check the msg size

test/ulog_parsing_test.cpp

@@ -266,6 +266,92 @@ TEST_CASE("ULog parsing - test corruption")
   }
 }
 
+TEST_CASE("ULog parsing - corruption recovery rejects wrong-size data for a known msg_id")
+{
+  // Same technique as the "test corruption" case above, but instead of only inserting zero
+  // bytes, follow them with a fabricated Data message header: known msg_type, plausible
+  // msg_size, and even the SAME msg_id as a real subscription - but with a payload size that
+  // doesn't match that subscription's format. Before the isValidDataMessage() fix,
+  // tryToRecover() would accept this as a valid resync point purely because it looks
+  // superficially plausible, and it would end up stored as a corrupted sample. The fix must
+  // reject it and instead recover at the next genuine message.
+  std::vector<uint8_t> injected_bytes;
+  std::vector<uint8_t> written_data;
+  TestWriter writer([&](const uint8_t* data, int length) {
+    if (!injected_bytes.empty()) {
+      written_data.insert(written_data.end(), injected_bytes.begin(), injected_bytes.end());
+      injected_bytes.clear();
+    }
+    const int prev_size = written_data.size();
+    written_data.resize(written_data.size() + length);
+    memcpy(written_data.data() + prev_size, data, length);
+  });
+
+  // A tiny, unpadded format (8 + 4 = 12 bytes), so its on-wire size is unambiguous.
+  const ulog_cpp::MessageFormat format{"target_message",
+                                       {{"uint64_t", "timestamp"}, {"float", "value"}}};
+  const uint16_t msg_id = 1;
+  const ulog_cpp::AddLoggedMessage add_logged_message{0, msg_id, "target_message"};
+
+  auto make_data = [&](uint64_t timestamp, float value) {
+    std::vector<uint8_t> bytes(12);
+    memcpy(bytes.data(), &timestamp, sizeof(timestamp));
+    memcpy(bytes.data() + sizeof(timestamp), &value, sizeof(value));
+    return ulog_cpp::Data{msg_id, bytes};
+  };
+  const auto data1 = make_data(100, 1.5F);
+  const auto data2 = make_data(200, 2.5F);
+
+  writer.fileHeader(ulog_cpp::FileHeader{});
+  writer.messageFormat(format);
+  writer.headerComplete();
+  writer.addLoggedMessage(add_logged_message);
+  writer.data(data1);
+
+  // Zero run to trigger corruptionDetected(), followed by the fabricated candidate.
+  injected_bytes.resize(50, 0);
+  const uint16_t fake_payload_size = 20;                 // target_message's real payload size is 12
+  const uint16_t fake_msg_size = fake_payload_size + 2;  // + 2-byte msg_id
+  auto append_u16 = [&](uint16_t v) {
+    injected_bytes.push_back(static_cast<uint8_t>(v & 0xFF));
+    injected_bytes.push_back(static_cast<uint8_t>((v >> 8) & 0xFF));
+  };
+  append_u16(fake_msg_size);
+  injected_bytes.push_back('D');  // ULogMessageType::DATA
+  append_u16(msg_id);             // the SAME msg_id as the real subscription above
+  // Fill the (wrong-size) payload with a byte value that can't be mistaken for any known
+  // msg_type character at any scan offset, so the search deterministically skips through it.
+  injected_bytes.resize(injected_bytes.size() + fake_payload_size, 0xAA);
+
+  writer.data(data2);  // flushes injected_bytes first, then the genuine next message
+
+  REQUIRE_GT(written_data.size(), 0);
+  REQUIRE_EQ(writer.num_errors, 0);
+
+  // Read it
+  const auto data_container =
+      std::make_shared<ulog_cpp::DataContainer>(ulog_cpp::DataContainer::StorageConfig::FullLog);
+  ulog_cpp::Reader reader{data_container};
+  // As in the "test corruption" case above: if recovery completes with no remaining
+  // external bytes to process (length == 0 at the point tryToRecover's recursive
+  // readChunk() call runs), the just-recovered message stays buffered but unprocessed
+  // until a later readChunk() call with length > 0 flushes it - readChunk()'s main loop
+  // only runs while length > 0. So a second, small trailing call is needed here too.
+  const int last_chunk_size = 5;
+  reader.readChunk(written_data.data(), written_data.size() - last_chunk_size);
+  reader.readChunk(written_data.data() + written_data.size() - last_chunk_size, last_chunk_size);
+
+  // Expected to have errors, but not be fatal
+  CHECK_GT(data_container->parsingErrors().size(), 0);
+  REQUIRE_FALSE(data_container->hadFatalError());
+
+  // The fabricated wrong-size candidate must NOT have been accepted as a third sample.
+  const auto& samples = data_container->subscriptionsByMessageId().at(msg_id)->rawSamples();
+  REQUIRE_EQ(samples.size(), 2);
+  CHECK_EQ(data1, samples[0]);
+  CHECK_EQ(data2, samples[1]);
+}
+
 struct MyData {
   uint64_t timestamp;
   float debug_array[4];

ulog_cpp/data_container.cpp

@@ -156,8 +156,38 @@ void DataContainer::data(const Data& data)
   if (iter == _subscriptions_by_message_id.end()) {
     throw ParsingException("Invalid subscription");
   }
+
+  // Guard against a Data message whose payload doesn't belong to this subscription's
+  // format at all. This can happen when the byte stream desyncs around a
+  // corrupted/dropped region: the framing (msg_size/msg_type/msg_id) can still look
+  // superficially valid while the payload actually belongs to a different message
+  // entirely. Without this check such a message would be silently decoded using the
+  // wrong field layout (reading garbage as e.g. a timestamp/float), rather than being
+  // discarded. Reader::tryToRecover() also calls isValidDataMessage() with this same
+  // logic before ever accepting such a candidate as a resync point in the first place -
+  // this check here is a defense-in-depth backstop for the (normally unreachable) case
+  // of a bad message slipping through outside of recovery.
+  if (!isValidDataMessage(data.msgId(), static_cast<uint16_t>(data.data().size()))) {
+    const auto& format = *iter->second->format();
+    throw ParsingException("Invalid data size for msg_id=" + std::to_string(data.msgId()) + " (" +
+                           iter->second->getAddLoggedMessage().messageName() + ") has size " +
+                           std::to_string(data.data().size()) + ", expected between " +
+                           std::to_string(format.minWireSizeBytes()) + " and " +
+                           std::to_string(format.sizeBytes()));
+  }
+
   iter->second->emplaceSample(std::move(data));
 }
+bool DataContainer::isValidDataMessage(uint16_t msg_id, uint16_t payload_size) const
+{
+  const auto iter = _subscriptions_by_message_id.find(msg_id);
+  if (iter == _subscriptions_by_message_id.end()) {
+    return false;
+  }
+  const auto& format = *iter->second->format();
+  const auto actual_size = static_cast<int>(payload_size);
+  return actual_size >= format.minWireSizeBytes() && actual_size <= format.sizeBytes();
+}
 void DataContainer::dropout(const Dropout& dropout)
 {
   if (_header_complete && _storage_config == StorageConfig::Header) {

ulog_cpp/data_container.hpp

@@ -62,6 +62,7 @@ class DataContainer : public DataHandlerInterface {
   void logging(const Logging& logging) override;
   void data(const Data& data) override;
   void dropout(const Dropout& dropout) override;
+  bool isValidDataMessage(uint16_t msg_id, uint16_t payload_size) const override;
 
   // Stored data
   bool isHeaderComplete() const { return _header_complete; }

ulog_cpp/data_handler_interface.hpp

@@ -28,6 +28,18 @@ class DataHandlerInterface {
   virtual void dropout(const Dropout& dropout) {}
   virtual void sync(const Sync& sync) {}
 
+  /**
+   * Used by Reader's corruption-recovery search to check a candidate byte offset that
+   * looks like it could be the start of a DATA message before accepting it as a resync
+   * point. Reader has no knowledge of subscriptions/formats itself, so it delegates the
+   * check here. Default is permissive (accepts everything).
+   * @param msg_id the candidate message's embedded msg_id
+   * @param payload_size the candidate message's payload size (msg_size minus the 2-byte
+   * msg_id)
+   * @return true if msg_id is a known subscription and payload_size is plausible for it
+   */
+  virtual bool isValidDataMessage(uint16_t msg_id, uint16_t payload_size) const { return true; }
+
  private:
 };
 

ulog_cpp/messages.cpp

@@ -430,6 +430,19 @@ int MessageFormat::sizeBytes() const
   return size;
 }
 
+int MessageFormat::minWireSizeBytes() const
+{
+  int end = static_cast<int>(_fields_ordered.size());
+  while (end > 0 && _fields_ordered[end - 1]->name().rfind("_padding", 0) == 0) {
+    --end;
+  }
+  int size = 0;
+  for (int i = 0; i < end; ++i) {
+    size += _fields_ordered[i]->sizeBytes();
+  }
+  return size;
+}
+
 void MessageFormat::resolveDefinition(
     const std::map<std::string, std::shared_ptr<MessageFormat>>& existing_formats) const
 {

ulog_cpp/messages.hpp

@@ -592,6 +592,21 @@ class MessageFormat {
    */
   int sizeBytes() const;
 
+  /**
+   * Returns the minimum size of this MessageFormat in bytes as it can appear on the wire in a
+   * Data message. This is only valid once the MessageFormat has been resolved (see sizeBytes()).
+   *
+   * ULog allows to omits trailing alignment padding fields (_padding0, etc.) from the
+   * on-wire Data payload, since they carry no information - the FORMAT definition still lists
+   * them because it describes the full in-memory struct layout (needed e.g. to correctly compute
+   * offsets when this format is used as a nested type in an array). sizeBytes() sums every field
+   * including such trailing padding, so it's an upper bound; this is the corresponding lower
+   * bound. A real Data message's payload size should fall within [minWireSizeBytes(),
+   * sizeBytes()].
+   * @return the minimum size of this MessageFormat on the wire, in bytes
+   */
+  int minWireSizeBytes() const;
+
   /**
    * @return the list of fields, in-order
    */

ulog_cpp/reader.cpp

@@ -207,8 +207,35 @@ void Reader::tryToRecover(const uint8_t* data, int length)
         if (header->msg_size != 0 && header->msg_type != 0 && header->msg_size < 10000 &&
             kKnownMessageTypes.find(static_cast<ULogMessageType>(header->msg_type)) !=
                 kKnownMessageTypes.end()) {
-          found = true;
-          break;
+          // A DATA candidate additionally has to reference a real subscription with a
+          // plausible payload size for it - the checks above alone (known type + size
+          // cap) are satisfied by a lot of unrelated byte patterns, and DATA messages
+          // are by far the most common type, so without this a resync can easily lock
+          // onto the wrong offset: bytes that happen to look like a Data header for some
+          // other message entirely, rather than the true next message. See
+          // DataHandlerInterface::isValidDataMessage().
+          bool candidate_valid = true;
+          if (static_cast<ULogMessageType>(header->msg_type) == ULogMessageType::DATA) {
+            if (header->msg_size < 2 || _partial_message_buffer_length - index <
+                                            static_cast<int>(sizeof(ulog_message_header_s)) + 2) {
+              // Too small to even hold the 2-byte msg_id, or not enough buffered data
+              // yet to see it - can't validate this candidate, so don't accept it
+              // (more data may still arrive for a later pass).
+              candidate_valid = false;
+            } else {
+              uint16_t candidate_msg_id = 0;
+              memcpy(&candidate_msg_id,
+                     _partial_message_buffer + index + sizeof(ulog_message_header_s),
+                     sizeof(candidate_msg_id));
+              const auto candidate_payload_size = static_cast<uint16_t>(header->msg_size - 2);
+              candidate_valid = _data_handler_interface->isValidDataMessage(candidate_msg_id,
+                                                                            candidate_payload_size);
+            }
+          }
+          if (candidate_valid) {
+            found = true;
+            break;
+          }
         }
       }