net: harden fragmented reply completion

An error array was read at index 0 without checking its size, so a packet
carrying "e":[] dereferenced a null element pointer and crashed the process
on unauthenticated UDP input.

When a message had an error array but no "a", "r" or "u" map, the same object
was then walked through the map view of the msgpack union. A two-element array
appeared to hold two key/value pairs, reading past its allocation. A request
field that was not a map was never type-checked either.

Validate the error array size, restrict request parsing to "a", "r" and "u",
and require those to be maps. Add parser regressions for the empty error
array, the error-only message and the non-map request; the first one
segfaults without this change.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

include/opendht/network_engine.h

@@ -489,8 +489,12 @@ private:
     static constexpr size_t MTU {1280};
     static constexpr size_t MAX_PACKET_VALUE_SIZE {600};
     static constexpr size_t MAX_MESSAGE_VALUE_SIZE {56 * 1024};
+    static constexpr size_t RX_MAX_EARLY_BYTES {MAX_MESSAGE_VALUE_SIZE + MAX_VALUE_SIZE + 32};
+    static constexpr size_t RX_MAX_EARLY_PARTS {RX_MAX_EARLY_BYTES / MTU + 2};
 
     void process(std::unique_ptr<ParsedMessage>&&, const SockAddr& from);
+    bool isExpectedReply(const ParsedMessage&, const SockAddr& from);
+    bool tryCompleteMessage(PartialMessage&, const SockAddr& from);
 
     bool rateLimit(const SockAddr& addr);
 
@@ -527,7 +531,8 @@ private:
     {
         return packValueHeader(buf, values.begin(), values.end());
     }
-    void maintainRxBuffer(Tid tid);
+    using PartialMessageKey = std::pair<SockAddr, Tid>;
+    void maintainRxBuffer(const PartialMessageKey& key);
 
     /*************
      *  Answers  *
@@ -573,7 +578,7 @@ private:
 
     // requests handling
     std::map<Tid, Sp<Request>> requests {};
-    std::map<Tid, PartialMessage> partial_messages;
+    std::map<PartialMessageKey, PartialMessage> partial_messages;
 
     MessageStats in_stats {}, out_stats {};
     std::set<SockAddr> blacklist {};

src/network_engine.cpp

@@ -40,8 +40,44 @@ struct NetworkEngine::PartialMessage
     time_point start;
     time_point last_part;
     std::unique_ptr<ParsedMessage> msg;
+    std::vector<std::unique_ptr<ParsedMessage>> early_parts;
+    size_t early_bytes {};
+    bool request_quota_consumed {};
 };
 
+bool
+NetworkEngine::isExpectedReply(const ParsedMessage& msg, const SockAddr& from)
+{
+    auto request = requests.find(msg.tid);
+    if (request != requests.end() && request->second->pending() && !request->second->node->id
+        && request->second->node->getAddr() == from)
+        return true;
+
+    auto node = cache.getNode(msg.id, from.getFamily());
+    if (!node || node->getAddr() != from)
+        return false;
+
+    auto pending = node->getRequest(msg.tid);
+    return (pending && pending->pending()) || node->getSocket(msg.tid);
+}
+
+bool
+NetworkEngine::tryCompleteMessage(PartialMessage& partial, const SockAddr& from)
+{
+    try {
+        if (!partial.msg->complete())
+            return false;
+        process(std::move(partial.msg), from);
+    } catch (const std::exception& e) {
+        if (logger_)
+            logger_->warn("Error while processing partial message: {}", e.what());
+    } catch (...) {
+        if (logger_)
+            logger_->warn("Unknown error while processing partial message");
+    }
+    return true;
+}
+
 std::vector<Blob>
 serializeValues(const std::vector<Sp<Value>>& st)
 {
@@ -466,35 +502,58 @@ NetworkEngine::processMessage(const uint8_t* buf, size_t buflen, SockAddr f)
 
     // partial value data
     if (msg->type == MessageType::ValueData) {
-        auto pmsg_it = partial_messages.find(msg->tid);
-        if (pmsg_it == partial_messages.end()) {
-            if (logIncoming_)
-                if (logger_)
-                    logger_->debug("Unable to find partial message");
+        if (msg->fragment_parts.empty()
+            || std::any_of(msg->fragment_parts.begin(), msg->fragment_parts.end(), [](const auto& fragment) {
+                   return fragment.second.second.empty();
+               })) {
             rateLimit(from);
             return;
         }
-        if (!pmsg_it->second.from.equals(from)) {
-            if (logger_)
-                logger_->debug("Received partial message data from unexpected IP address");
-            rateLimit(from);
+        PartialMessageKey key {from, msg->tid};
+        auto pmsg_it = partial_messages.find(key);
+        if (pmsg_it == partial_messages.end()) {
+            if (!rateLimit(from))
+                return;
+            auto inserted = partial_messages.emplace(key, PartialMessage {});
+            pmsg_it = inserted.first;
+            pmsg_it->second.from = from;
+            pmsg_it->second.start = now;
+            pmsg_it->second.last_part = now;
+            pmsg_it->second.request_quota_consumed = true;
+            scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, key));
+        }
+        auto& pmsg = pmsg_it->second;
+        if (!pmsg.msg) {
+            if (pmsg.early_parts.size() >= RX_MAX_EARLY_PARTS) {
+                partial_messages.erase(pmsg_it);
+                return;
+            }
+            size_t fragment_bytes {};
+            for (const auto& [index, fragment] : msg->fragment_parts) {
+                (void) index;
+                if (fragment.second.size() > RX_MAX_EARLY_BYTES - fragment_bytes) {
+                    partial_messages.erase(pmsg_it);
+                    return;
+                }
+                fragment_bytes += fragment.second.size();
+            }
+            if (fragment_bytes > RX_MAX_EARLY_BYTES - pmsg.early_bytes) {
+                partial_messages.erase(pmsg_it);
+                return;
+            }
+            pmsg.early_bytes += fragment_bytes;
+            pmsg.early_parts.emplace_back(std::move(msg));
+            pmsg.last_part = now;
+            scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, key));
             return;
         }
         // append data block
-        if (pmsg_it->second.msg->append(*msg)) {
-            pmsg_it->second.last_part = now;
-            // check data completion
-            if (pmsg_it->second.msg->complete()) {
-                try {
-                    // process the full message
-                    process(std::move(pmsg_it->second.msg), from);
-                } catch (const std::exception& e) {
-                    if (logger_)
-                        logger_->warn("Error while processing partial message: {}", e.what());
-                }
+        if (pmsg.msg->append(*msg)) {
+            pmsg.last_part = now;
+            if (tryCompleteMessage(pmsg, from)) {
                 partial_messages.erase(pmsg_it);
             } else
-                scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, msg->tid));
+                scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, key));
         }
         return;
     }
@@ -505,7 +564,32 @@ NetworkEngine::processMessage(const uint8_t* buf, size_t buflen, SockAddr f)
         return;
     }
 
-    if (msg->type > MessageType::Reply) {
+    if (msg->value_parts.empty()) {
+        if (msg->type > MessageType::Reply) {
+            /* Rate limit requests. */
+            if (!rateLimit(from)) {
+                if (logger_)
+                    logger_->warn("Dropping request due to rate limiting");
+                return;
+            }
+        }
+        try {
+            process(std::move(msg), from);
+        } catch (const std::exception& e) {
+            if (logger_)
+                logger_->warn("Error while processing message: {}", e.what());
+        }
+        return;
+    }
+
+    PartialMessageKey partial_key {from, msg->tid};
+    if (msg->type == MessageType::Reply && !isExpectedReply(*msg, from))
+        return;
+
+    const auto partial = partial_messages.find(partial_key);
+    const bool request_quota_consumed = partial != partial_messages.end() && !partial->second.msg
+                                        && partial->second.request_quota_consumed;
+    if (msg->type > MessageType::Reply && !request_quota_consumed) {
         /* Rate limit requests. */
         if (!rateLimit(from)) {
             if (logger_)
@@ -514,28 +598,28 @@ NetworkEngine::processMessage(const uint8_t* buf, size_t buflen, SockAddr f)
         }
     }
 
-    if (msg->value_parts.empty()) {
-        try {
-            process(std::move(msg), from);
-        } catch (const std::exception& e) {
-            if (logger_)
-                logger_->warn("Error while processing message: {}", e.what());
-            return;
-        }
-    } else {
-        // starting partial message session
-        auto k = msg->tid;
-        auto& pmsg = partial_messages[k];
-        if (not pmsg.msg) {
+    // starting partial message session
+    auto inserted = partial_messages.emplace(partial_key, PartialMessage {});
+    auto& pmsg = inserted.first->second;
+    if (not pmsg.msg) {
+        if (inserted.second) {
             pmsg.from = from;
-            pmsg.msg = std::move(msg);
             pmsg.start = now;
-            pmsg.last_part = now;
-            scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, k));
-            scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, k));
-        } else if (logger_)
-            logger_->error("Partial message with given TID {} already exists", k);
-    }
+            scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, partial_key));
+        }
+        pmsg.msg = std::move(msg);
+        pmsg.last_part = now;
+        for (const auto& part : pmsg.early_parts)
+            pmsg.msg->append(*part);
+        pmsg.early_parts.clear();
+        pmsg.early_bytes = 0;
+        if (tryCompleteMessage(pmsg, from)) {
+            partial_messages.erase(inserted.first);
+        } else {
+            scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, partial_key));
+        }
+    } else if (logger_)
+        logger_->error("Partial message from {} with given TID {} already exists", from.toString(), msg->tid);
 }
 
 void
@@ -1614,9 +1698,9 @@ NetworkEngine::sendError(const SockAddr& addr, Tid tid, uint16_t code, const std
 }
 
 void
-NetworkEngine::maintainRxBuffer(Tid tid)
+NetworkEngine::maintainRxBuffer(const PartialMessageKey& key)
 {
-    auto msg = partial_messages.find(tid);
+    auto msg = partial_messages.find(key);
     if (msg != partial_messages.end()) {
         const auto& now = scheduler.time();
         if (msg->second.start + RX_MAX_PACKET_TIME < now || msg->second.last_part + RX_TIMEOUT < now) {

src/parsed_message.h

@@ -274,14 +274,21 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
 
     if (!parsed.a && !parsed.r && !parsed.e && !parsed.u)
         throw msgpack::type_error();
-    auto& req = parsed.a ? *parsed.a : (parsed.r ? *parsed.r : (parsed.u ? *parsed.u : *parsed.e));
 
     if (parsed.e) {
-        if (parsed.e->type != msgpack::type::ARRAY)
+        if (parsed.e->type != msgpack::type::ARRAY or parsed.e->via.array.size < 1)
             throw msgpack::type_error();
         error_code = parsed.e->via.array.ptr[0].as<uint16_t>();
     }
 
+    /* An error-only message carries no request map. The error array must never
+       be reached through the map view of the msgpack object union, and a
+       request field that is not a map must be rejected rather than
+       reinterpreted. */
+    const msgpack::object* req = parsed.a ? parsed.a : (parsed.r ? parsed.r : parsed.u);
+    if (req and req->type != msgpack::type::MAP)
+        throw msgpack::type_error();
+
     struct ParsedReq
     {
         msgpack::object* values;
@@ -290,8 +297,9 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
         msgpack::object* want;
     } parsedReq {};
 
-    for (unsigned i = 0; i < req.via.map.size; i++) {
-        auto& o = req.via.map.ptr[i];
+    const uint32_t req_size = req ? req->via.map.size : 0;
+    for (unsigned i = 0; i < req_size; i++) {
+        auto& o = req->via.map.ptr[i];
         if (o.key.type != msgpack::type::STR)
             continue;
         auto key = o.key.as<std::string_view>();
@@ -370,6 +378,8 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
     } else if (parsedReq.fields) {
         if (auto rfields = findMapValue(*parsedReq.fields, "f"sv)) {
             auto vfields = rfields->as<std::set<Value::Field>>();
+            if (vfields.empty())
+                throw msgpack::type_error();
             if (auto rvalues = findMapValue(*parsedReq.fields, "v"sv)) {
                 if (rvalues->type != msgpack::type::ARRAY)
                     throw msgpack::type_error();

tests/test_networkengine.cpp

@@ -214,9 +214,12 @@ makeEngine(std::unique_ptr<TestDatagramSocket>&& socket,
            std::mt19937_64& rd,
            int& onNewNodeCalls)
 {
+    net::NetworkConfig config {};
+    config.max_req_per_sec = -1;
+    config.max_peer_req_per_sec = -1;
     return net::NetworkEngine(
         myid,
-        {},
+        config,
         std::move(socket),
         {},
         rd,
@@ -234,6 +237,21 @@ makeEngine(std::unique_ptr<TestDatagramSocket>&& socket,
         [](Sp<Node>, const InfoHash&, const Blob&, const Value::Id&) { return net::RequestAnswer {}; });
 }
 
+static Sp<Node>
+addPendingGetRequest(net::NetworkEngine& engine, const InfoHash& id, const SockAddr& addr, Tid tid)
+{
+    auto node = engine.insertNode(id, addr);
+    auto request = std::make_shared<net::Request>(
+        MessageType::GetValues,
+        tid,
+        node,
+        Blob {},
+        [](const net::Request&, ParsedMessage&&) {},
+        [](const net::Request&, bool) {});
+    node->requested(request);
+    return node;
+}
+
 void
 NetworkEngineTester::setUp()
 {}
@@ -243,24 +261,130 @@ NetworkEngineTester::tearDown()
 {}
 
 void
-NetworkEngineTester::testIgnoresUnknownPartialData()
+NetworkEngineTester::testCompletesPartialSessionWithDataBeforeHeader()
 {
     Scheduler scheduler;
     std::mt19937_64 rd(0);
     InfoHash myid = InfoHash::getRandom(rd);
+    InfoHash remoteId = InfoHash::getRandom(rd);
     int onNewNodeCalls = 0;
     auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
 
-    Blob fragment = {1, 2, 3};
-    auto packet = makeValueDataPacketBlob(99, 0, 0, fragment);
+    auto serialized = serializeValue("fragment before header");
     auto from = makeIPv4("127.0.0.2", 5000);
+    addPendingGetRequest(engine, remoteId, from, 99);
+    auto initialOnNewNodeCalls = onNewNodeCalls;
+    auto fragment = makeValueDataPacketBlob(99, 0, 0, serialized);
+    auto header = makeReplyHeaderPacketBlob(remoteId, 99, {serialized.size()});
 
+    engine.processMessage(fragment.data(), fragment.size(), from);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
+    CPPUNIT_ASSERT_EQUAL(initialOnNewNodeCalls, onNewNodeCalls);
+
+    engine.processMessage(header.data(), header.size(), from);
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
+    CPPUNIT_ASSERT(onNewNodeCalls > initialOnNewNodeCalls);
+}
+
+void
+NetworkEngineTester::testBuffersLargeValueBeforeHeader()
+{
+    Scheduler scheduler;
+    std::mt19937_64 rd(8);
+    InfoHash myid = InfoHash::getRandom(rd);
+    InfoHash remoteId = InfoHash::getRandom(rd);
+    int onNewNodeCalls = 0;
+    auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
+
+    auto serialized = serializeValue(std::string(57 * 1024, 'l'));
+    auto from = makeIPv4("127.0.0.2", 5008);
+    addPendingGetRequest(engine, remoteId, from, 101);
+    auto initialOnNewNodeCalls = onNewNodeCalls;
+    for (size_t offset = 0; offset < serialized.size(); offset += TEST_MTU) {
+        auto end = std::min(offset + TEST_MTU, serialized.size());
+        Blob data(serialized.begin() + offset, serialized.begin() + end);
+        auto fragment = makeValueDataPacketBlob(101, 0, offset, data);
+        engine.processMessage(fragment.data(), fragment.size(), from);
+    }
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
+
+    auto header = makeReplyHeaderPacketBlob(remoteId, 101, {serialized.size()});
+    engine.processMessage(header.data(), header.size(), from);
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
+    CPPUNIT_ASSERT(onNewNodeCalls > initialOnNewNodeCalls);
+}
+
+void
+NetworkEngineTester::testIgnoresEmptyPartialData()
+{
+    Scheduler scheduler;
+    std::mt19937_64 rd(7);
+    InfoHash myid = InfoHash::getRandom(rd);
+    int onNewNodeCalls = 0;
+    auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
+
+    auto packet = makeValueDataPacketBlob(100, 0, 0, {});
+    auto from = makeIPv4("127.0.0.2", 5007);
     engine.processMessage(packet.data(), packet.size(), from);
 
     CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
     CPPUNIT_ASSERT_EQUAL(0, onNewNodeCalls);
 }
 
+void
+NetworkEngineTester::testRejectsUnsolicitedPartialReply()
+{
+    Scheduler scheduler;
+    std::mt19937_64 rd(9);
+    InfoHash myid = InfoHash::getRandom(rd);
+    InfoHash remoteId = InfoHash::getRandom(rd);
+    int onNewNodeCalls = 0;
+    auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
+
+    auto from = makeIPv4("127.0.0.2", 5009);
+    auto header = makeReplyHeaderPacketBlob(remoteId, 102, {TEST_MTU});
+    engine.processMessage(header.data(), header.size(), from);
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getNodeCacheSize());
+}
+
+void
+NetworkEngineTester::testDropsMalformedCompletedPartialMessage()
+{
+    Scheduler scheduler;
+    std::mt19937_64 rd(10);
+    InfoHash myid = InfoHash::getRandom(rd);
+    InfoHash remoteId = InfoHash::getRandom(rd);
+    int onNewNodeCalls = 0;
+    auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
+
+    auto from = makeIPv4("127.0.0.2", 5010);
+    addPendingGetRequest(engine, remoteId, from, 103);
+    Blob malformed {0xc1};
+    auto header = makeReplyHeaderPacketBlob(remoteId, 103, {malformed.size()});
+    auto fragment = makeValueDataPacketBlob(103, 0, 0, malformed);
+
+    engine.processMessage(header.data(), header.size(), from);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
+    engine.processMessage(fragment.data(), fragment.size(), from);
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
+
+    auto earlyFrom = makeIPv4("127.0.0.2", 5011);
+    addPendingGetRequest(engine, remoteId, earlyFrom, 104);
+    auto earlyHeader = makeReplyHeaderPacketBlob(remoteId, 104, {malformed.size()});
+    auto earlyFragment = makeValueDataPacketBlob(104, 0, 0, malformed);
+
+    engine.processMessage(earlyFragment.data(), earlyFragment.size(), earlyFrom);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
+    engine.processMessage(earlyHeader.data(), earlyHeader.size(), earlyFrom);
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
+}
+
 void
 NetworkEngineTester::testCompletesPartialSessionFromSameSource()
 {
@@ -274,6 +398,8 @@ NetworkEngineTester::testCompletesPartialSessionFromSameSource()
     std::string data(TEST_MTU * 2 + 64, 'z');
     auto serialized = serializeValue(data);
     auto from = makeIPv4("127.0.0.2", 5001);
+    addPendingGetRequest(engine, remoteId, from, 7);
+    auto initialOnNewNodeCalls = onNewNodeCalls;
     auto header = makeReplyHeaderPacketBlob(remoteId, 7, {serialized.size()});
 
     engine.processMessage(header.data(), header.size(), from);
@@ -287,45 +413,40 @@ NetworkEngineTester::testCompletesPartialSessionFromSameSource()
     }
 
     CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-    CPPUNIT_ASSERT(onNewNodeCalls > 0);
+    CPPUNIT_ASSERT(onNewNodeCalls > initialOnNewNodeCalls);
 }
 
 void
-NetworkEngineTester::testKeepsSessionForWrongSourceFragment()
+NetworkEngineTester::testSeparatesPartialSessionsBySource()
 {
     Scheduler scheduler;
     std::mt19937_64 rd(2);
     InfoHash myid = InfoHash::getRandom(rd);
-    InfoHash remoteId = InfoHash::getRandom(rd);
+    InfoHash firstRemoteId = InfoHash::getRandom(rd);
+    InfoHash secondRemoteId = InfoHash::getRandom(rd);
     int onNewNodeCalls = 0;
     auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
 
-    std::string data(TEST_MTU + 32, 'w');
-    auto serialized = serializeValue(data);
-    auto goodFrom = makeIPv4("127.0.0.2", 5002);
-    auto badFrom = makeIPv4("127.0.0.3", 5002);
-    auto header = makeReplyHeaderPacketBlob(remoteId, 11, {serialized.size()});
+    auto serialized = serializeValue("same transaction, different peers");
+    auto firstFrom = makeIPv4("127.0.0.2", 5002);
+    auto secondFrom = makeIPv4("127.0.0.3", 5002);
+    addPendingGetRequest(engine, firstRemoteId, firstFrom, 11);
+    addPendingGetRequest(engine, secondRemoteId, secondFrom, 11);
+    auto initialOnNewNodeCalls = onNewNodeCalls;
+    auto firstHeader = makeReplyHeaderPacketBlob(firstRemoteId, 11, {serialized.size()});
+    auto secondHeader = makeReplyHeaderPacketBlob(secondRemoteId, 11, {serialized.size()});
 
-    engine.processMessage(header.data(), header.size(), goodFrom);
+    engine.processMessage(firstHeader.data(), firstHeader.size(), firstFrom);
+    engine.processMessage(secondHeader.data(), secondHeader.size(), secondFrom);
+    CPPUNIT_ASSERT_EQUAL((size_t) 2, engine.getPartialCount());
+
+    auto fragment = makeValueDataPacketBlob(11, 0, 0, serialized);
+    engine.processMessage(fragment.data(), fragment.size(), firstFrom);
     CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
-
-    auto firstEnd = std::min(TEST_MTU, serialized.size());
-    Blob firstFragment(serialized.begin(), serialized.begin() + firstEnd);
-    auto firstPacket = makeValueDataPacketBlob(11, 0, 0, firstFragment);
-    engine.processMessage(firstPacket.data(), firstPacket.size(), badFrom);
-
-    CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
-    CPPUNIT_ASSERT_EQUAL(0, onNewNodeCalls);
-
-    engine.processMessage(firstPacket.data(), firstPacket.size(), goodFrom);
-    if (firstEnd < serialized.size()) {
-        Blob tail(serialized.begin() + firstEnd, serialized.end());
-        auto tailPacket = makeValueDataPacketBlob(11, 0, firstEnd, tail);
-        engine.processMessage(tailPacket.data(), tailPacket.size(), goodFrom);
-    }
+    engine.processMessage(fragment.data(), fragment.size(), secondFrom);
 
     CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-    CPPUNIT_ASSERT(onNewNodeCalls > 0);
+    CPPUNIT_ASSERT(onNewNodeCalls > initialOnNewNodeCalls);
 }
 
 void

tests/test_networkengine.h

@@ -13,9 +13,13 @@ class NetworkEngineTester : public CppUnit::TestFixture
 #ifdef _MSC_VER
     CPPUNIT_TEST(testDisabledOnMsvc);
 #else
-    CPPUNIT_TEST(testIgnoresUnknownPartialData);
+    CPPUNIT_TEST(testCompletesPartialSessionWithDataBeforeHeader);
+    CPPUNIT_TEST(testBuffersLargeValueBeforeHeader);
+    CPPUNIT_TEST(testIgnoresEmptyPartialData);
+    CPPUNIT_TEST(testRejectsUnsolicitedPartialReply);
+    CPPUNIT_TEST(testDropsMalformedCompletedPartialMessage);
     CPPUNIT_TEST(testCompletesPartialSessionFromSameSource);
-    CPPUNIT_TEST(testKeepsSessionForWrongSourceFragment);
+    CPPUNIT_TEST(testSeparatesPartialSessionsBySource);
     CPPUNIT_TEST(testListenConfirmationCarriesToken);
     CPPUNIT_TEST(testListenConfirmationUpdatesSearchNodeToken);
     CPPUNIT_TEST(testListenReopensSocketAfterNodeExpiration);
@@ -30,9 +34,13 @@ public:
 #ifdef _MSC_VER
     void testDisabledOnMsvc();
 #else
-    void testIgnoresUnknownPartialData();
+    void testCompletesPartialSessionWithDataBeforeHeader();
+    void testBuffersLargeValueBeforeHeader();
+    void testIgnoresEmptyPartialData();
+    void testRejectsUnsolicitedPartialReply();
+    void testDropsMalformedCompletedPartialMessage();
     void testCompletesPartialSessionFromSameSource();
-    void testKeepsSessionForWrongSourceFragment();
+    void testSeparatesPartialSessionsBySource();
     void testListenConfirmationCarriesToken();
     void testListenConfirmationUpdatesSearchNodeToken();
     void testListenReopensSocketAfterNodeExpiration();

tests/test_parsedmessage.cpp

@@ -328,6 +328,104 @@ ParsedMessageTester::testParseRejectsInvalidValuesField()
     CPPUNIT_ASSERT_THROW(parsed.msgpack_unpack(msg.get()), msgpack::type_error);
 }
 
+void
+ParsedMessageTester::testParseRejectsEmptyFieldSelection()
+{
+    msgpack::sbuffer buffer;
+    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+
+    pk.pack_map(3);
+    pk.pack(KEY_R);
+    pk.pack_map(1);
+    pk.pack(KEY_REQ_FIELDS);
+    pk.pack_map(2);
+    pk.pack("f");
+    pk.pack_array(0);
+    pk.pack("v");
+    pk.pack_array(0);
+    pk.pack(KEY_TID);
+    pk.pack(1u);
+    pk.pack(KEY_Y);
+    pk.pack(KEY_R);
+
+    auto msg = msgpack::unpack(buffer.data(), buffer.size());
+    ParsedMessage parsed;
+    CPPUNIT_ASSERT_THROW(parsed.msgpack_unpack(msg.get()), msgpack::type_error);
+}
+
+void
+ParsedMessageTester::testParseRejectsEmptyErrorArray()
+{
+    msgpack::sbuffer buffer;
+    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+
+    pk.pack_map(3);
+    pk.pack(KEY_E);
+    pk.pack_array(0);
+    pk.pack(KEY_TID);
+    pk.pack(1u);
+    pk.pack(KEY_Y);
+    pk.pack(KEY_E);
+
+    auto msg = msgpack::unpack(buffer.data(), buffer.size());
+    ParsedMessage parsed;
+    CPPUNIT_ASSERT_THROW(parsed.msgpack_unpack(msg.get()), msgpack::type_error);
+}
+
+void
+ParsedMessageTester::testParseRejectsNonMapRequest()
+{
+    msgpack::sbuffer buffer;
+    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+
+    pk.pack_map(4);
+    pk.pack(KEY_A);
+    pk.pack_array(2);
+    pk.pack(1u);
+    pk.pack(2u);
+    pk.pack(KEY_Q);
+    pk.pack(QUERY_PING);
+    pk.pack(KEY_TID);
+    pk.pack(1u);
+    pk.pack(KEY_Y);
+    pk.pack("q");
+
+    auto msg = msgpack::unpack(buffer.data(), buffer.size());
+    ParsedMessage parsed;
+    CPPUNIT_ASSERT_THROW(parsed.msgpack_unpack(msg.get()), msgpack::type_error);
+}
+
+void
+ParsedMessageTester::testParseErrorOnlyMessageKeepsRequestFieldsEmpty()
+{
+    msgpack::sbuffer buffer;
+    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+
+    // An error message with no request map: the error array must not be walked
+    // as if it were a sequence of key/value pairs.
+    pk.pack_map(3);
+    pk.pack(KEY_E);
+    pk.pack_array(2);
+    pk.pack(401u);
+    pk.pack("unauthorized");
+    pk.pack(KEY_TID);
+    pk.pack(9u);
+    pk.pack(KEY_Y);
+    pk.pack(KEY_E);
+
+    auto msg = msgpack::unpack(buffer.data(), buffer.size());
+    ParsedMessage parsed;
+    parsed.msgpack_unpack(msg.get());
+
+    CPPUNIT_ASSERT(parsed.type == MessageType::Error);
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 401, parsed.error_code);
+    CPPUNIT_ASSERT(not parsed.id);
+    CPPUNIT_ASSERT(parsed.token.empty());
+    CPPUNIT_ASSERT(parsed.values.empty());
+    CPPUNIT_ASSERT(parsed.nodes4_raw.empty());
+    CPPUNIT_ASSERT(parsed.nodes6_raw.empty());
+}
+
 void
 ParsedMessageTester::testParseIgnoresIncompleteValueDataEntry()
 {

tests/test_parsedmessage.h

@@ -24,6 +24,10 @@ class ParsedMessageTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testParseRejectsNonMapPacket);
     CPPUNIT_TEST(testParseRejectsInvalidValueDataPayload);
     CPPUNIT_TEST(testParseRejectsInvalidValuesField);
+    CPPUNIT_TEST(testParseRejectsEmptyFieldSelection);
+    CPPUNIT_TEST(testParseRejectsEmptyErrorArray);
+    CPPUNIT_TEST(testParseRejectsNonMapRequest);
+    CPPUNIT_TEST(testParseErrorOnlyMessageKeepsRequestFieldsEmpty);
     CPPUNIT_TEST(testParseIgnoresIncompleteValueDataEntry);
     CPPUNIT_TEST(testCompleteEmpty);
     CPPUNIT_TEST(testCompleteIncomplete);
@@ -53,6 +57,10 @@ public:
     void testParseRejectsNonMapPacket();
     void testParseRejectsInvalidValueDataPayload();
     void testParseRejectsInvalidValuesField();
+    void testParseRejectsEmptyFieldSelection();
+    void testParseRejectsEmptyErrorArray();
+    void testParseRejectsNonMapRequest();
+    void testParseErrorOnlyMessageKeepsRequestFieldsEmpty();
     void testParseIgnoresIncompleteValueDataEntry();
     void testCompleteEmpty();
     void testCompleteIncomplete();