Fix CVE-2025-62799

This commit fixes CVE-2025-62799 by refactoring the handling of RTPS DATA_FRAG messages and
reserving space for possible alignment required to store a fragment index.

This is an squashed commit of a privately reviewed branch.

Signed-off-by: Miguel Company <miguelcompany@eprosima.com>
Reviewed-by: cferreiragonz <carlosferreira@eprosima.com>

Author: MiguelCompany

include/fastdds/rtps/common/CacheChange.hpp

@@ -21,6 +21,7 @@
 
 #include <atomic>
 #include <cassert>
+#include <limits>
 
 #include <fastdds/rtps/common/ChangeKind_t.hpp>
 #include <fastdds/rtps/common/FragmentNumber.hpp>
@@ -328,6 +329,62 @@ struct FASTDDS_EXPORTED_API CacheChange_t
         return is_fully_assembled();
     }
 
+    /**
+     * @brief Calculate the minimum required payload size to store a fragmented change.
+     *
+     * @param[in]  payload_size       Size of the full payload.
+     * @param[in]  fragment_size      Size of each fragment.
+     * @param[out] min_required_size  Minimum required size to store the fragmented payload.
+     */
+    static bool calculate_required_fragmented_payload_size(
+            uint32_t payload_size,
+            uint16_t fragment_size,
+            uint32_t& min_required_size)
+    {
+        if ((0 == fragment_size) || (payload_size <= fragment_size))
+        {
+            min_required_size = payload_size;
+            return true;
+        }
+
+        // In order to avoid overflow on the calculations, we limit the maximum payload size
+        constexpr uint32_t MAX_PAYLOAD_SIZE = std::numeric_limits<uint32_t>::max() - 4u - 3u;
+        if (payload_size > MAX_PAYLOAD_SIZE)
+        {
+            return false;
+        }
+
+        // Ensure fragment size is at least 4 bytes to store fragment index
+        if (fragment_size < 4u)
+        {
+            return false;
+        }
+
+        // Calculate number of fragments without risk of overflow
+        uint32_t fragment_count = payload_size / fragment_size;
+        if (0 != (payload_size % fragment_size))
+        {
+            ++fragment_count;
+        }
+
+        // This cannot overflow as the result will always be <= payload_size
+        uint32_t last_fragment_offset = (fragment_count - 1) * fragment_size;
+
+        // Since we will write a fragment index at the beginning of each fragment,
+        // we need to ensure there is space for it in the last fragment.
+        // Note: we already imposed limits to ensure no overflow occurs.
+        min_required_size = (last_fragment_offset + 3u) & ~3u; // Align last fragment size to 4 bytes
+        min_required_size += 4u; // Add fragment index size
+
+        // Ensure minimum size is at least payload size
+        if (min_required_size < payload_size)
+        {
+            min_required_size = payload_size;
+        }
+
+        return true;
+    }
+
 private:
 
     // Fragment size

src/cpp/rtps/builtin/discovery/endpoint/EDPSimple.cpp

@@ -277,7 +277,7 @@ void EDPSimple::processPersistentData(
 
                 CacheChange_t* change_to_add = nullptr;
 
-                if (!reader.first->reserve_cache(change->serializedPayload.length, change_to_add)) //Reserve a new cache from the corresponding cache pool
+                if (!reader.first->reserve_cache(change->serializedPayload.length, 0, change_to_add)) //Reserve a new cache from the corresponding cache pool
                 {
                     EPROSIMA_LOG_ERROR(RTPS_EDP, "Problem reserving CacheChange in EDPServer reader");
                     return;

src/cpp/rtps/builtin/discovery/participant/PDPServer.cpp

@@ -1033,7 +1033,7 @@ bool PDPServer::remove_remote_participant(
 
         // TODO check in standard if DROP payload is always 0
         // We create the drop from Reader to make release simplier
-        endpoints->reader.reader_->reserve_cache(mp_builtin->m_att.writerPayloadSize, pC);
+        endpoints->reader.reader_->reserve_cache(mp_builtin->m_att.writerPayloadSize, 0, pC);
 
         // We must create the corresponding DATA(p[UD])
         if (nullptr != pC)
@@ -1718,7 +1718,7 @@ bool PDPServer::process_backup_discovery_database_restore(
             std::istringstream(it.value()["change"]["sample_identity"].get<std::string>()) >> sample_identity_aux;
 
             // Reserve memory for new change. There will not be changes from own server
-            if (!endpoints->reader.reader_->reserve_cache(length, change_aux))
+            if (!endpoints->reader.reader_->reserve_cache(length, 0, change_aux))
             {
                 EPROSIMA_LOG_ERROR(RTPS_PDP_SERVER, "Error creating CacheChange");
                 // TODO release changes and exit
@@ -1757,7 +1757,7 @@ bool PDPServer::process_backup_discovery_database_restore(
             else
             {
                 // Reserve memory for new change. There will not be changes from own server
-                if (!edp->publications_reader_.first->reserve_cache(length, change_aux))
+                if (!edp->publications_reader_.first->reserve_cache(length, 0, change_aux))
                 {
                     EPROSIMA_LOG_ERROR(RTPS_PDP_SERVER, "Error creating CacheChange");
                     // TODO release changes and exit
@@ -1796,7 +1796,7 @@ bool PDPServer::process_backup_discovery_database_restore(
             else
             {
                 // Reserve memory for new change. There will not be changes from own server
-                if (!edp->subscriptions_reader_.first->reserve_cache(length, change_aux))
+                if (!edp->subscriptions_reader_.first->reserve_cache(length, 0, change_aux))
                 {
                     EPROSIMA_LOG_ERROR(RTPS_PDP_SERVER, "Error creating CacheChange");
                     // TODO release changes and exit

src/cpp/rtps/reader/BaseReader.cpp

@@ -18,6 +18,7 @@
 
 #include <rtps/reader/BaseReader.hpp>
 
+#include <algorithm>
 #include <cassert>
 #include <cstdint>
 #include <mutex>
@@ -279,21 +280,45 @@ std::shared_ptr<LocalReaderPointer> BaseReader::get_local_pointer()
 
 bool BaseReader::reserve_cache(
         uint32_t cdr_payload_size,
+        uint16_t fragment_size,
         fastdds::rtps::CacheChange_t*& change)
 {
     std::lock_guard<decltype(mp_mutex)> guard(mp_mutex);
 
     change = nullptr;
 
+    // Calculate and validate required payload size
+    uint32_t reserve_size = fixed_payload_size_ > 0 ? fixed_payload_size_ : cdr_payload_size;
+    uint32_t payload_size = std::min(reserve_size, cdr_payload_size);
+    uint32_t min_required_size = 0;
+    if (!CacheChange_t::calculate_required_fragmented_payload_size(payload_size, fragment_size, min_required_size))
+    {
+        EPROSIMA_LOG_WARNING(RTPS_READER,
+                "Required payload size calculation overflows for payload size '" << payload_size <<
+                "' and fragment size '" << fragment_size << "'");
+        return false;
+    }
+
+    if (min_required_size > reserve_size)
+    {
+        if (fixed_payload_size_ > 0)
+        {
+            EPROSIMA_LOG_WARNING(RTPS_READER,
+                    "Fixed payload size '" << fixed_payload_size_ <<
+                    "' is insufficient for fragmentation with fragment size '" << fragment_size << "'");
+            return false;
+        }
+        reserve_size = min_required_size;
+    }
+
     fastdds::rtps::CacheChange_t* reserved_change = nullptr;
     if (!change_pool_->reserve_cache(reserved_change))
     {
         EPROSIMA_LOG_WARNING(RTPS_READER, "Problem reserving cache from pool");
         return false;
     }
 
-    uint32_t payload_size = fixed_payload_size_ ? fixed_payload_size_ : cdr_payload_size;
-    if (!payload_pool_->get_payload(payload_size, reserved_change->serializedPayload))
+    if (!payload_pool_->get_payload(reserve_size, reserved_change->serializedPayload))
     {
         change_pool_->release_cache(reserved_change);
         EPROSIMA_LOG_WARNING(RTPS_READER, "Problem reserving payload from pool");

src/cpp/rtps/reader/BaseReader.hpp

@@ -177,12 +177,14 @@ class BaseReader
      * @brief Reserve a CacheChange_t.
      *
      * @param [in]  cdr_payload_size  Size of the received payload.
+     * @param [in]  fragment_size     Size of each fragment (0 if not fragmented).
      * @param [out] change            Pointer to the reserved change.
      *
      * @return True if correctly reserved.
      */
     bool reserve_cache(
             uint32_t cdr_payload_size,
+            uint16_t fragment_size,
             fastdds::rtps::CacheChange_t*& change);
 
     /**

src/cpp/rtps/reader/StatefulReader.cpp

@@ -738,7 +738,7 @@ bool StatefulReader::process_data_frag_msg(
             if (!history_->get_change(change_to_add->sequenceNumber, change_to_add->writerGUID, &work_change))
             {
                 // A new change should be reserved
-                if (reserve_cache(sampleSize, work_change))
+                if (reserve_cache(sampleSize, change_to_add->getFragmentSize(), work_change))
                 {
                     if (work_change->serializedPayload.max_size < sampleSize)
                     {

src/cpp/rtps/reader/StatelessReader.cpp

@@ -793,7 +793,7 @@ bool StatelessReader::process_data_frag_msg(
                 // Check if a new change should be reserved
                 if (work_change == nullptr)
                 {
-                    if (reserve_cache(sampleSize, work_change))
+                    if (reserve_cache(sampleSize, change_to_add->getFragmentSize(), work_change))
                     {
                         if (work_change->serializedPayload.max_size < sampleSize)
                         {

test/blackbox/common/BlackboxTestsTransportUDP.cpp

@@ -808,6 +808,103 @@ TEST(TransportUDP, KeyOnlyBigPayloadIgnored_BestEffort)
     KeyOnlyBigPayloadIgnored(writer, reader);
 }
 
+// Regression test for redmine issue #23830
+TEST(TransportUDP, MaliciousDataFragUnalignedSizes)
+{
+    // Force using UDP transport
+    auto udp_transport = std::make_shared<UDPv4TransportDescriptor>();
+
+    PubSubWriter<UnboundedHelloWorldPubSubType> writer(TEST_TOPIC_NAME);
+    PubSubReader<UnboundedHelloWorldPubSubType> reader(TEST_TOPIC_NAME);
+
+    struct MaliciousDataFragUnalignedSizes
+    {
+        std::array<char, 4> rtps_id{ {'R', 'T', 'P', 'S'} };
+        std::array<uint8_t, 2> protocol_version{ {2, 3} };
+        std::array<uint8_t, 2> vendor_id{ {0x01, 0x0F} };
+        GuidPrefix_t sender_prefix{};
+
+        struct DataFragSubMsg
+        {
+            struct Header
+            {
+                uint8_t submessage_id = 0x16;
+#if FASTDDS_IS_BIG_ENDIAN_TARGET
+                uint8_t flags = 0x00;
+#else
+                uint8_t flags = 0x01;
+#endif  // FASTDDS_IS_BIG_ENDIAN_TARGET
+                uint16_t octets_to_next_header = 0x22;
+                uint16_t extra_flags = 0;
+                uint16_t octets_to_inline_qos = 0x1c;
+                EntityId_t reader_id{};
+                EntityId_t writer_id{};
+                SequenceNumber_t sn{100};
+                uint32_t fragment_starting_num = 2;
+                uint16_t fragments_in_submessage = 1;
+                uint16_t fragment_size = 50002;
+                uint32_t sample_size = 50003;
+            };
+
+            struct SerializedData
+            {
+                octet data[2] {0xAA, 0xAA};
+            };
+
+            Header header;
+            SerializedData payload;
+        }
+        data;
+    };
+
+    UDPMessageSender fake_msg_sender;
+
+    // Set common QoS
+    reader.disable_builtin_transport().add_user_transport_to_pparams(udp_transport)
+            .history_depth(10).reliability(eprosima::fastdds::dds::RELIABLE_RELIABILITY_QOS);
+    writer.history_depth(10).reliability(eprosima::fastdds::dds::RELIABLE_RELIABILITY_QOS);
+
+    // Set custom reader locator so we can send malicious data to a known location
+    Locator_t reader_locator;
+    ASSERT_TRUE(IPLocator::setIPv4(reader_locator, "127.0.0.1"));
+    reader_locator.port = 7000;
+    reader.add_to_unicast_locator_list("127.0.0.1", 7000);
+
+    // Initialize and wait for discovery
+    reader.init();
+    ASSERT_TRUE(reader.isInitialized());
+    writer.init();
+    ASSERT_TRUE(writer.isInitialized());
+
+    reader.wait_discovery();
+    writer.wait_discovery();
+
+    auto data = default_unbounded_helloworld_data_generator();
+    reader.startReception(data);
+    writer.send(data);
+    ASSERT_TRUE(data.empty());
+
+    // Send malicious data
+    {
+        auto writer_guid = writer.datawriter_guid();
+
+        MaliciousDataFragUnalignedSizes malicious_packet{};
+        malicious_packet.sender_prefix = writer_guid.guidPrefix;
+        malicious_packet.data.header.writer_id = writer_guid.entityId;
+        malicious_packet.data.header.reader_id = reader.datareader_guid().entityId;
+
+        CDRMessage_t msg(0);
+        uint32_t msg_len = static_cast<uint32_t>(sizeof(malicious_packet));
+        msg.init(reinterpret_cast<octet*>(&malicious_packet), msg_len);
+        msg.length = msg_len;
+        msg.pos = msg_len;
+        fake_msg_sender.send(msg, reader_locator);
+    }
+
+    // Block reader until reception finished or timeout.
+    reader.block_for_all();
+}
+
 // Test for ==operator UDPTransportDescriptor is not required as it is an abstract class and in UDPv4 is same method
 // Test for copy UDPTransportDescriptor is not required as it is an abstract class and in UDPv4 is same method
 

test/unittest/rtps/common/CacheChangeTests.cpp

@@ -142,6 +142,85 @@ TEST(CacheChange, FragmentManagement)
     }
 }
 
+TEST(CacheChange, calculate_required_fragmented_payload_size)
+{
+    struct TestCase
+    {
+        uint32_t payload_size;
+        uint16_t fragment_size;
+        bool expected_result;
+        uint32_t expected_min_required_size;
+    };
+
+    const std::vector<TestCase> test_cases =
+    {
+        // No fragmentation
+        {100u, 0u, true, 100u},
+        {50u, 100u, true, 50u},
+        // Minimum fragment size
+        {100u, 1u, false, 0u},
+        {100u, 2u, false, 0u},
+        {100u, 3u, false, 0u},
+        // Fragmentation cases
+        {100u, 20u, true, 100u},
+        {101u, 20u, true, 104u},
+        {102u, 20u, true, 104u},
+        {103u, 20u, true, 104u},
+        {50004u, 50003u, true, 50008u},
+        {50003u, 50002u, true, 50008u},
+        {50002u, 50001u, true, 50008u},
+        {50001u, 50000u, true, 50004u},
+        // Typical UDP max fragment size
+        {1234567u, 64000u, true, 1234567u},
+        {1234568u, 64000u, true, 1234568u},
+        {1234569u, 64000u, true, 1234569u},
+        {1234570u, 64000u, true, 1234570u},
+        // Edge cases
+        {5u, 4u, true, 8u},
+        {6u, 4u, true, 8u},
+        {7u, 4u, true, 8u},
+        {8u, 4u, true, 8u},
+        {
+            std::numeric_limits<uint32_t>::max() - 4u - 3u,
+            std::numeric_limits<uint16_t>::max(),
+            true,
+            std::numeric_limits<uint32_t>::max() - 4u - 3u
+        },
+        {
+            std::numeric_limits<uint32_t>::max() - 4u - 3u,
+            4u,
+            true,
+            std::numeric_limits<uint32_t>::max() - 4u - 3u
+        }
+    };
+
+    for (const auto& test_case : test_cases)
+    {
+        uint32_t min_required_size = 0;
+        bool result = CacheChange_t::calculate_required_fragmented_payload_size(
+            test_case.payload_size,
+            test_case.fragment_size,
+            min_required_size);
+        EXPECT_EQ(result, test_case.expected_result)
+            << "Failed for payload_size=" << test_case.payload_size
+            << ", fragment_size=" << test_case.fragment_size;
+        if (result)
+        {
+            EXPECT_EQ(min_required_size, test_case.expected_min_required_size)
+                << "Failed for payload_size=" << test_case.payload_size
+                << ", fragment_size=" << test_case.fragment_size;
+        }
+    }
+
+    // Oversized payload
+    constexpr uint32_t MAX_PAYLOAD_SIZE = std::numeric_limits<uint32_t>::max() - 4u - 3u;
+    for (uint32_t payload_size = std::numeric_limits<uint32_t>::max(); payload_size > MAX_PAYLOAD_SIZE; --payload_size)
+    {
+        uint32_t min_required_size = 0;
+        EXPECT_FALSE(CacheChange_t::calculate_required_fragmented_payload_size(payload_size, 20, min_required_size));
+    }
+}
+
 int main(
         int argc,
         char** argv)