udp: properly handle truncated/dropped datagrams (envoyproxy#14122)

Signed-off-by: Matt Klein <mklein@lyft.com>
Signed-off-by: Christoph Pakulski <christoph@tetrate.io>
Co-authored-by: Matt Klein <mklein@lyft.com>
Co-authored-by: Christoph Pakulski <christoph@tetrate.io>
Signed-off-by: Christoph Pakulski <christoph@tetrate.io>

docs/root/intro/version_history.rst

@@ -5,6 +5,7 @@ Version history
 ================
 Changes
 -------
+* udp: fixed issue in which receiving truncated UDP datagrams would cause Envoy to crash.
 
 1.13.6 (September 29, 2020)
 ===========================

include/envoy/network/io_handle.h

@@ -4,6 +4,8 @@
 #include "envoy/common/platform.h"
 #include "envoy/common/pure.h"
 
+#include "absl/container/fixed_array.h"
+
 namespace Envoy {
 namespace Buffer {
 struct RawSlice;
@@ -74,13 +76,26 @@ class IoHandle {
                                           int flags, const Address::Ip* self_ip,
                                           const Address::Instance& peer_address) PURE;
 
+  struct RecvMsgPerPacketInfo {
+    // If true indicates a successful syscall, but the packet was dropped due to truncation. We do
+    // not support receiving truncated packets.
+    bool truncated_and_dropped_{false};
+  };
+
+  /**
+   * The output parameter type for recvmsg and recvmmsg.
+   */
   struct RecvMsgOutput {
     /*
+     * @param num_packets_per_call is the max number of packets allowed per
+     * recvmmsg call. For recvmsg call, any value larger than 0 is allowed, but
+     * only one packet will be returned.
      * @param dropped_packets points to a variable to store how many packets are
      * dropped so far. If nullptr, recvmsg() won't try to get this information
      * from transport header.
      */
-    RecvMsgOutput(uint32_t* dropped_packets) : dropped_packets_(dropped_packets) {}
+    RecvMsgOutput(size_t num_packets_per_call, uint32_t* dropped_packets)
+        : dropped_packets_(dropped_packets), msg_(num_packets_per_call) {}
 
     // If not nullptr, its value is the total number of packets dropped. recvmsg() will update it
     // when more packets are dropped.
@@ -89,6 +104,8 @@ class IoHandle {
     std::shared_ptr<const Address::Instance> local_address_;
     // The the source address from transport header.
     std::shared_ptr<const Address::Instance> peer_address_;
+    // Per packet info.
+    absl::FixedArray<RecvMsgPerPacketInfo> msg_;
   };
 
   /**

source/common/network/io_socket_handle_impl.cc

@@ -16,6 +16,21 @@ using Envoy::Api::SysCallIntResult;
 using Envoy::Api::SysCallSizeResult;
 
 namespace Envoy {
+
+namespace {
+
+constexpr int messageTruncatedOption() {
+#if defined(__APPLE__)
+  // OSX does not support passing `MSG_TRUNC` to recvmsg and recvmmsg. This does not effect
+  // functionality and it primarily used for logging.
+  return 0;
+#else
+  return MSG_TRUNC;
+#endif
+}
+
+} // namespace
+
 namespace Network {
 
 IoSocketHandleImpl::~IoSocketHandleImpl() {
@@ -243,11 +258,18 @@ Api::IoCallUint64Result IoSocketHandleImpl::recvmsg(Buffer::RawSlice* slices,
   cmsg->cmsg_len = cmsg_space;
   hdr.msg_control = cmsg;
   hdr.msg_controllen = cmsg_space;
-  auto& os_sys_calls = Api::OsSysCallsSingleton::get();
-  const Api::SysCallSizeResult result = os_sys_calls.recvmsg(fd_, &hdr, 0);
+  Api::SysCallSizeResult result =
+      Api::OsSysCallsSingleton::get().recvmsg(fd_, &hdr, messageTruncatedOption());
   if (result.rc_ < 0) {
     return sysCallResultToIoCallResult(result);
   }
+  if ((hdr.msg_flags & MSG_TRUNC) != 0) {
+    ENVOY_LOG_MISC(debug, "Dropping truncated UDP packet with size: {}.", result.rc_);
+    result.rc_ = 0;
+    (*output.dropped_packets_)++;
+    output.msg_[0].truncated_and_dropped_ = true;
+    return sysCallResultToIoCallResult(result);
+  }
 
   RELEASE_ASSERT((hdr.msg_flags & MSG_CTRUNC) == 0,
                  fmt::format("Incorrectly set control message length: {}", hdr.msg_controllen));
@@ -287,7 +309,8 @@ Api::IoCallUint64Result IoSocketHandleImpl::recvmsg(Buffer::RawSlice* slices,
       if (output.dropped_packets_ != nullptr) {
         absl::optional<uint32_t> maybe_dropped = maybeGetPacketsDroppedFromHeader(*cmsg);
         if (maybe_dropped) {
-          *output.dropped_packets_ = *maybe_dropped;
+          *output.dropped_packets_ += *maybe_dropped;
+          continue;
         }
       }
     }

source/common/network/udp_listener_impl.h

@@ -24,8 +24,8 @@ class UdpListenerImpl : public BaseListenerImpl,
 public:
   UdpListenerImpl(Event::DispatcherImpl& dispatcher, SocketSharedPtr socket,
                   UdpListenerCallbacks& cb, TimeSource& time_source);
-
   ~UdpListenerImpl() override;
+  uint32_t packetsDropped() { return packets_dropped_; }
 
   // Network::Listener Interface
   void disable() override;

source/common/network/utility.cc

@@ -536,11 +536,11 @@ Api::IoCallUint64Result Utility::readFromSocket(IoHandle& handle,
   const uint64_t num_slices = buffer->reserve(udp_packet_processor.maxPacketSize(), &slice, 1);
   ASSERT(num_slices == 1);
 
-  IoHandle::RecvMsgOutput output(packets_dropped);
+  IoHandle::RecvMsgOutput output(1, packets_dropped);
   Api::IoCallUint64Result result =
       handle.recvmsg(&slice, num_slices, local_address.ip()->port(), output);
 
-  if (!result.ok()) {
+  if (!result.ok() || output.msg_[0].truncated_and_dropped_) {
     return result;
   }
 
@@ -581,12 +581,6 @@ Api::IoErrorPtr Utility::readPacketsFromSocket(IoHandle& handle,
       return std::move(result.err_);
     }
 
-    if (result.rc_ == 0) {
-      // TODO(conqerAtapple): Is zero length packet interesting? If so add stats
-      // for it. Otherwise remove the warning log below.
-      ENVOY_LOG_MISC(trace, "received 0-length packet");
-    }
-
     if (packets_dropped != old_packets_dropped) {
       // The kernel tracks SO_RXQ_OVFL as a uint32 which can overflow to a smaller
       // value. So as long as this count differs from previously recorded value,

test/common/network/udp_listener_impl_test.cc

@@ -94,7 +94,7 @@ class UdpListenerImplTest : public ListenerImplTestBase {
   SocketSharedPtr server_socket_;
   SocketPtr client_socket_;
   Address::InstanceConstSharedPtr send_to_addr_;
-  MockUdpListenerCallbacks listener_callbacks_;
+  NiceMock<MockUdpListenerCallbacks> listener_callbacks_;
   std::unique_ptr<UdpListenerImpl> listener_;
 };
 
@@ -172,6 +172,82 @@ TEST_P(UdpListenerImplTest, UseActualDstUdp) {
   dispatcher_->run(Event::Dispatcher::RunType::Block);
 }
 
+// Test a large datagram that gets dropped using recvmmsg if supported.
+TEST_P(UdpListenerImplTest, LargeDatagramRecvmmsg) {
+  client_socket_ = createClientSocket(false);
+
+  // This will get dropped.
+  const std::string first(4096, 'a');
+  const void* void_pointer = static_cast<const void*>(first.c_str());
+  Buffer::RawSlice first_slice{const_cast<void*>(void_pointer), first.length()};
+  const std::string second("second");
+  void_pointer = static_cast<const void*>(second.c_str());
+  Buffer::RawSlice second_slice{const_cast<void*>(void_pointer), second.length()};
+  // This will get dropped.
+  const std::string third(4096, 'b');
+  void_pointer = static_cast<const void*>(third.c_str());
+  Buffer::RawSlice third_slice{const_cast<void*>(void_pointer), third.length()};
+
+  auto send_rc = Network::Utility::writeToSocket(client_socket_->ioHandle(), &first_slice, 1,
+                                                 nullptr, *send_to_addr_);
+  ASSERT_EQ(send_rc.rc_, first.length());
+  send_rc = Network::Utility::writeToSocket(client_socket_->ioHandle(), &second_slice, 1, nullptr,
+                                            *send_to_addr_);
+  ASSERT_EQ(send_rc.rc_, second.length());
+  send_rc = Network::Utility::writeToSocket(client_socket_->ioHandle(), &third_slice, 1, nullptr,
+                                            *send_to_addr_);
+  ASSERT_EQ(send_rc.rc_, third.length());
+
+  EXPECT_CALL(listener_callbacks_, onReadReady());
+  EXPECT_CALL(listener_callbacks_, onData(_)).WillOnce(Invoke([&](const UdpRecvData& data) -> void {
+    validateRecvCallbackParams(data);
+    EXPECT_EQ(data.buffer_->toString(), second);
+
+    dispatcher_->exit();
+  }));
+
+  dispatcher_->run(Event::Dispatcher::RunType::Block);
+  EXPECT_EQ(2, listener_->packetsDropped());
+}
+
+// Test a large datagram that gets dropped using recvmsg.
+TEST_P(UdpListenerImplTest, LargeDatagramRecvmsg) {
+  client_socket_ = createClientSocket(false);
+
+  // This will get dropped.
+  const std::string first(4096, 'a');
+  const void* void_pointer = static_cast<const void*>(first.c_str());
+  Buffer::RawSlice first_slice{const_cast<void*>(void_pointer), first.length()};
+  const std::string second("second");
+  void_pointer = static_cast<const void*>(second.c_str());
+  Buffer::RawSlice second_slice{const_cast<void*>(void_pointer), second.length()};
+  // This will get dropped.
+  const std::string third(4096, 'b');
+  void_pointer = static_cast<const void*>(third.c_str());
+  Buffer::RawSlice third_slice{const_cast<void*>(void_pointer), third.length()};
+
+  auto send_rc = Network::Utility::writeToSocket(client_socket_->ioHandle(), &first_slice, 1,
+                                                 nullptr, *send_to_addr_);
+  ASSERT_EQ(send_rc.rc_, first.length());
+  send_rc = Network::Utility::writeToSocket(client_socket_->ioHandle(), &second_slice, 1, nullptr,
+                                            *send_to_addr_);
+  ASSERT_EQ(send_rc.rc_, second.length());
+  send_rc = Network::Utility::writeToSocket(client_socket_->ioHandle(), &third_slice, 1, nullptr,
+                                            *send_to_addr_);
+  ASSERT_EQ(send_rc.rc_, third.length());
+
+  EXPECT_CALL(listener_callbacks_, onReadReady());
+  EXPECT_CALL(listener_callbacks_, onData(_)).WillOnce(Invoke([&](const UdpRecvData& data) -> void {
+    validateRecvCallbackParams(data);
+    EXPECT_EQ(data.buffer_->toString(), second);
+
+    dispatcher_->exit();
+  }));
+
+  dispatcher_->run(Event::Dispatcher::RunType::Block);
+  EXPECT_EQ(2, listener_->packetsDropped());
+}
+
 /**
  * Tests UDP listener for read and write callbacks with actual data.
  */

test/extensions/quic_listeners/quiche/quic_io_handle_wrapper_test.cc

@@ -1,5 +1,6 @@
 #include <sys/socket.h>
 
+#include <cstddef>
 #include <memory>
 
 #include "common/network/address_impl.h"
@@ -56,8 +57,8 @@ TEST_F(QuicIoHandleWrapperTest, DelegateIoHandleCalls) {
   EXPECT_CALL(os_sys_calls_, sendmsg(fd, _, 0)).WillOnce(Return(Api::SysCallSizeResult{5u, 0}));
   wrapper_->sendmsg(&slice, 1, 0, /*self_ip=*/nullptr, *addr);
 
-  Network::IoHandle::RecvMsgOutput output(nullptr);
-  EXPECT_CALL(os_sys_calls_, recvmsg(fd, _, 0)).WillOnce(Invoke([](int, struct msghdr* msg, int) {
+  Network::IoHandle::RecvMsgOutput output(1, nullptr);
+  EXPECT_CALL(os_sys_calls_, recvmsg(fd, _, MSG_TRUNC)).WillOnce(Invoke([](int, msghdr* msg, int) {
     sockaddr_storage ss;
     auto ipv6_addr = reinterpret_cast<sockaddr_in6*>(&ss);
     memset(ipv6_addr, 0, sizeof(sockaddr_in6));
@@ -66,6 +67,7 @@ TEST_F(QuicIoHandleWrapperTest, DelegateIoHandleCalls) {
     ipv6_addr->sin6_port = htons(54321);
     *reinterpret_cast<sockaddr_in6*>(msg->msg_name) = *ipv6_addr;
     msg->msg_namelen = sizeof(sockaddr_in6);
+    msg->msg_controllen = 0;
     return Api::SysCallSizeResult{5u, 0};
   }));
   wrapper_->recvmsg(&slice, 1, /*self_port=*/12345, output);

test/test_common/threadsafe_singleton_injector.h

@@ -12,6 +12,7 @@ template <class T> class TestThreadsafeSingletonInjector {
     ThreadSafeSingleton<T>::instance_ = instance;
   }
   ~TestThreadsafeSingletonInjector() { ThreadSafeSingleton<T>::instance_ = latched_instance_; }
+  T& latched() { return *latched_instance_; }
 
 private:
   T* latched_instance_;

tools/spelling_dictionary.txt

@@ -285,6 +285,7 @@ TLV
 TMPDIR
 TODO
 TPM
+TRUNC
 TSAN
 TSI
 TTL