diff --git a/test/ulog_parsing_test.cpp b/test/ulog_parsing_test.cpp index 570a7bd..72b79ba 100644 --- a/test/ulog_parsing_test.cpp +++ b/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 injected_bytes; + std::vector 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 bytes(12); + memcpy(bytes.data(), ×tamp, 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(v & 0xFF)); + injected_bytes.push_back(static_cast((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::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]; diff --git a/ulog_cpp/data_container.cpp b/ulog_cpp/data_container.cpp index 39a10d2..3893150 100644 --- a/ulog_cpp/data_container.cpp +++ b/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(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(payload_size); + return actual_size >= format.minWireSizeBytes() && actual_size <= format.sizeBytes(); +} void DataContainer::dropout(const Dropout& dropout) { if (_header_complete && _storage_config == StorageConfig::Header) { diff --git a/ulog_cpp/data_container.hpp b/ulog_cpp/data_container.hpp index 961ba1b..e8e288e 100644 --- a/ulog_cpp/data_container.hpp +++ b/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; } diff --git a/ulog_cpp/data_handler_interface.hpp b/ulog_cpp/data_handler_interface.hpp index f503062..bd2e33f 100644 --- a/ulog_cpp/data_handler_interface.hpp +++ b/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: }; diff --git a/ulog_cpp/messages.cpp b/ulog_cpp/messages.cpp index 0a3ea8d..0b26712 100644 --- a/ulog_cpp/messages.cpp +++ b/ulog_cpp/messages.cpp @@ -430,6 +430,19 @@ int MessageFormat::sizeBytes() const return size; } +int MessageFormat::minWireSizeBytes() const +{ + int end = static_cast(_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>& existing_formats) const { diff --git a/ulog_cpp/messages.hpp b/ulog_cpp/messages.hpp index ce4b068..9d8e896 100644 --- a/ulog_cpp/messages.hpp +++ b/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 */ diff --git a/ulog_cpp/reader.cpp b/ulog_cpp/reader.cpp index 0470e0a..8a9bcfe 100644 --- a/ulog_cpp/reader.cpp +++ b/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(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(header->msg_type) == ULogMessageType::DATA) { + if (header->msg_size < 2 || _partial_message_buffer_length - index < + static_cast(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(header->msg_size - 2); + candidate_valid = _data_handler_interface->isValidDataMessage(candidate_msg_id, + candidate_payload_size); + } + } + if (candidate_valid) { + found = true; + break; + } } }