fix warning

include/xrpl/server/detail/UDPDoor.h

@@ -29,6 +29,7 @@
 #include <boost/asio/buffer.hpp>
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/post.hpp>
 #include <boost/asio/spawn.hpp>
 #include <boost/beast/core/detect_ssl.hpp>
 #include <boost/beast/core/multi_buffer.hpp>
@@ -60,6 +61,9 @@ private:
     std::vector<char> recv_buffer_;
     endpoint_type local_endpoint_;  // Store TCP-style endpoint
 
+    // Large enough for one maximum-size datagram (see do_send).
+    static constexpr int minSendBufferSize = 256 * 1024;
+
 public:
     UDPDoor(
         Handler& handler,
@@ -97,6 +101,27 @@ public:
             return;
         }
 
+        // Some platforms (macOS: net.inet.udp.maxdgram = 9216) cap outgoing
+        // datagrams at SO_SNDBUF, so a full-size fragment fails with
+        // EMSGSIZE unless the send buffer is at least that large.
+        {
+            boost::asio::socket_base::send_buffer_size current;
+            socket_.get_option(current, ec);
+            if (!ec && current.value() < minSendBufferSize)
+            {
+                socket_.set_option(
+                    boost::asio::socket_base::send_buffer_size(
+                        minSendBufferSize),
+                    ec);
+                if (ec)
+                {
+                    JLOG(j_.warn())
+                        << "UDP set send_buffer_size failed: " << ec.message();
+                }
+            }
+            ec.clear();
+        }
+
         socket_.bind(udp_endpoint, ec);
         if (ec)
         {
@@ -174,22 +199,50 @@ private:
             std::string(recv_buffer_.data(), bytes_transferred),
             tcp_endpoint,
             port_,
-            [this, tcp_endpoint](std::string const& response) {
-                do_send(response, tcp_endpoint);
+            [self = this->shared_from_this(),
+             tcp_endpoint](std::string const& response) {
+                self->do_send(response, tcp_endpoint);
             });
 
         do_receive();
     }
 
+    // The payload must stay alive until the send completes, and the socket
+    // is only touched on the strand.
+    void
+    send_datagram(
+        std::shared_ptr<std::string const> payload,
+        boost::asio::ip::udp::endpoint const& udp_endpoint)
+    {
+        boost::asio::post(
+            strand_,
+            [this,
+             self = this->shared_from_this(),
+             payload = std::move(payload),
+             udp_endpoint]() {
+                if (!socket_.is_open())
+                    return;
+
+                socket_.async_send_to(
+                    boost::asio::buffer(*payload),
+                    udp_endpoint,
+                    boost::asio::bind_executor(
+                        strand_,
+                        [self, payload](
+                            error_code ec, std::size_t bytes_transferred) {
+                            if (ec &&
+                                ec != boost::asio::error::operation_aborted)
+                            {
+                                JLOG(self->j_.error())
+                                    << "UDP send failed: " << ec.message();
+                            }
+                        }));
+            });
+    }
+
     void
     do_send(std::string const& response, endpoint_type const& tcp_endpoint)
     {
-        if (!socket_.is_open())
-        {
-            std::cout << "UDP SOCKET NOT OPEN WHEN SENDING\n\n";
-            return;
-        }
-
         const size_t HEADER_SIZE = 16;
         const size_t MAX_DATAGRAM_SIZE =
             65487;  // Allow for ipv6 header 40 bytes + 8 bytes of udp header
@@ -202,19 +255,8 @@ private:
         // If message fits in single datagram, send normally
         if (response.length() <= MAX_DATAGRAM_SIZE)
         {
-            socket_.async_send_to(
-                boost::asio::buffer(response),
-                udp_endpoint,
-                boost::asio::bind_executor(
-                    strand_,
-                    [this, self = this->shared_from_this()](
-                        error_code ec, std::size_t bytes_transferred) {
-                        if (ec && ec != boost::asio::error::operation_aborted)
-                        {
-                            JLOG(j_.error())
-                                << "UDP send failed: " << ec.message();
-                        }
-                    }));
+            send_datagram(
+                std::make_shared<std::string const>(response), udp_endpoint);
             return;
         }
 
@@ -233,51 +275,39 @@ private:
         // Send fragmented packets
         for (uint16_t packet_num = 0; packet_num < total_packets; packet_num++)
         {
-            std::string fragment;
-            fragment.reserve(MAX_DATAGRAM_SIZE);
+            auto fragment = std::make_shared<std::string>();
+            fragment->reserve(MAX_DATAGRAM_SIZE);
 
             // Add header - 4 bytes of zeros
-            fragment.push_back(0);
-            fragment.push_back(0);
-            fragment.push_back(0);
-            fragment.push_back(0);
+            fragment->push_back(0);
+            fragment->push_back(0);
+            fragment->push_back(0);
+            fragment->push_back(0);
 
             // Add packet number (little endian)
-            fragment.push_back(packet_num & 0xFF);
-            fragment.push_back((packet_num >> 8) & 0xFF);
+            fragment->push_back(packet_num & 0xFF);
+            fragment->push_back((packet_num >> 8) & 0xFF);
 
             // Add total packets (little endian)
-            fragment.push_back(total_packets & 0xFF);
-            fragment.push_back((total_packets >> 8) & 0xFF);
+            fragment->push_back(total_packets & 0xFF);
+            fragment->push_back((total_packets >> 8) & 0xFF);
 
             // Add timestamp (8 bytes, little endian)
-            fragment.push_back(timestamp & 0xFF);
-            fragment.push_back((timestamp >> 8) & 0xFF);
-            fragment.push_back((timestamp >> 16) & 0xFF);
-            fragment.push_back((timestamp >> 24) & 0xFF);
-            fragment.push_back((timestamp >> 32) & 0xFF);
-            fragment.push_back((timestamp >> 40) & 0xFF);
-            fragment.push_back((timestamp >> 48) & 0xFF);
-            fragment.push_back((timestamp >> 56) & 0xFF);
+            fragment->push_back(timestamp & 0xFF);
+            fragment->push_back((timestamp >> 8) & 0xFF);
+            fragment->push_back((timestamp >> 16) & 0xFF);
+            fragment->push_back((timestamp >> 24) & 0xFF);
+            fragment->push_back((timestamp >> 32) & 0xFF);
+            fragment->push_back((timestamp >> 40) & 0xFF);
+            fragment->push_back((timestamp >> 48) & 0xFF);
+            fragment->push_back((timestamp >> 56) & 0xFF);
 
             // Calculate payload slice
             size_t start = packet_num * payload_size;
             size_t length = std::min(payload_size, response.length() - start);
-            fragment.append(response.substr(start, length));
+            fragment->append(response.substr(start, length));
 
-            socket_.async_send_to(
-                boost::asio::buffer(fragment),
-                udp_endpoint,
-                boost::asio::bind_executor(
-                    strand_,
-                    [this, self = this->shared_from_this()](
-                        error_code ec, std::size_t bytes_transferred) {
-                        if (ec && ec != boost::asio::error::operation_aborted)
-                        {
-                            JLOG(j_.error())
-                                << "UDP send failed: " << ec.message();
-                        }
-                    }));
+            send_datagram(std::move(fragment), udp_endpoint);
         }
     }
 

src/test/server/Server_test.cpp

@@ -37,6 +37,8 @@
 
 #include <chrono>
 #include <condition_variable>
+#include <cstdint>
+#include <map>
 #include <mutex>
 #include <optional>
 #include <stdexcept>
@@ -498,6 +500,151 @@ public:
         run(false);
     }
 
+    // Answers every datagram through sendResponse: "big" gets a reply too
+    // large for one datagram, anything else gets "pong".
+    struct UDPEchoHandler : TestHandler
+    {
+        // Two fragments: small enough not to overflow a default socket buffer.
+        static constexpr std::size_t bigSize = 100000;
+
+        void
+        onUDPMessage(
+            std::string const& message,
+            boost::asio::ip::tcp::endpoint const& remoteEndpoint,
+            Port const& port,
+            std::function<void(std::string const&)> sendResponse)
+        {
+            if (message == "big")
+            {
+                std::string big(bigSize, '\0');
+                for (std::size_t i = 0; i < big.size(); ++i)
+                    big[i] = static_cast<char>('a' + (i % 26));
+                sendResponse(big);
+            }
+            else
+            {
+                sendResponse("pong");
+            }
+        }
+
+        static std::string
+        expectedBig()
+        {
+            std::string big(bigSize, '\0');
+            for (std::size_t i = 0; i < big.size(); ++i)
+                big[i] = static_cast<char>('a' + (i % 26));
+            return big;
+        }
+    };
+
+    // Exercises UDPDoor::do_send / send_datagram: a short reply goes out as
+    // one datagram, a long one is split into headered fragments.
+    void
+    udpResponseTests()
+    {
+        testcase("UDP RPC door responses");
+        using namespace std::chrono_literals;
+        using udp = boost::asio::ip::udp;
+
+        TestSink sink{*this};
+        TestThread thread;
+        sink.threshold(beast::severities::Severity::kAll);
+        beast::Journal journal{sink};
+
+        auto const addr = boost::asio::ip::make_address(getEnvLocalhostAddr());
+
+        // UDPDoor reports the configured port, not the bound one, so pick a
+        // free port up front.
+        std::uint16_t port = 0;
+        {
+            boost::asio::io_context io;
+            udp::socket probe(io, udp::endpoint(addr, 0));
+            port = probe.local_endpoint().port();
+        }
+
+        UDPEchoHandler handler;
+        auto s = make_Server(handler, thread.get_io_context(), journal);
+
+        std::vector<Port> ports(1);
+        ports.back().name = "udp_rpc";
+        ports.back().ip = addr;
+        ports.back().port = port;
+        ports.back().protocol.insert("udp");
+        s->ports(ports);
+
+        boost::asio::io_context io;
+        udp::socket client(io, udp::endpoint(addr, 0));
+        // Room for the full fragmented reply on platforms with small
+        // default socket buffers.
+        client.set_option(udp::socket::receive_buffer_size(512 * 1024));
+        udp::endpoint const server(addr, port);
+
+        // Receive one datagram, or nothing if the timeout expires.
+        auto receive = [&](std::chrono::milliseconds timeout)
+            -> std::optional<std::string> {
+            std::vector<char> buf(70000);
+            udp::endpoint from;
+            std::optional<std::string> result;
+            client.async_receive_from(
+                boost::asio::buffer(buf),
+                from,
+                [&](boost::system::error_code ec, std::size_t n) {
+                    if (!ec)
+                        result.emplace(buf.data(), n);
+                });
+            io.restart();
+            io.run_for(timeout);
+            if (!result)
+            {
+                client.cancel();
+                io.restart();
+                io.run();
+            }
+            return result;
+        };
+
+        // Single datagram reply
+        client.send_to(boost::asio::buffer(std::string("ping")), server);
+        if (auto const r = receive(5s); BEAST_EXPECT(r))
+            BEAST_EXPECT(*r == "pong");
+
+        // Fragmented reply
+        client.send_to(boost::asio::buffer(std::string("big")), server);
+        std::size_t constexpr headerSize = 16;
+        std::size_t constexpr payloadSize = 65487 - headerSize;
+        std::size_t const expectedPackets =
+            (UDPEchoHandler::bigSize + payloadSize - 1) / payloadSize;
+
+        std::map<std::uint16_t, std::string> fragments;
+        std::uint16_t total = 0;
+        while (fragments.size() < expectedPackets)
+        {
+            auto const r = receive(5s);
+            if (!BEAST_EXPECT(r) || !BEAST_EXPECT(r->size() > headerSize))
+                break;
+            auto const byte = [&](std::size_t i) {
+                return static_cast<std::uint16_t>(
+                    static_cast<unsigned char>((*r)[i]));
+            };
+            BEAST_EXPECT(byte(0) == 0 && byte(1) == 0);
+            BEAST_EXPECT(byte(2) == 0 && byte(3) == 0);
+            std::uint16_t const num = byte(4) | (byte(5) << 8);
+            total = byte(6) | (byte(7) << 8);
+            fragments[num] = r->substr(headerSize);
+        }
+
+        BEAST_EXPECT(total == expectedPackets);
+        if (BEAST_EXPECT(fragments.size() == expectedPackets))
+        {
+            std::string reassembled;
+            for (auto const& [num, data] : fragments)
+                reassembled += data;
+            BEAST_EXPECT(reassembled == UDPEchoHandler::expectedBig());
+        }
+
+        s = nullptr;
+    }
+
     void
     testBadConfig()
     {
@@ -642,6 +789,7 @@ public:
         basicTests();
         stressTest();
         udpHighwayTests();
+        udpResponseTests();
         testBadConfig();
     }
 };

src/xrpld/rpc/detail/ServerHandler.cpp

@@ -1126,6 +1126,10 @@ ServerHandler::onUDPMessage(
     if (port.has_peer())
         return app_.overlay().processXUSH(message, remoteEndpoint);
 
+    if (!ipAllowed(
+            remoteEndpoint.address(), port.admin_nets_v4, port.admin_nets_v6))
+        return;
+
     Json::Value jv;
     if (message.size() > RPC::Tuning::maxRequestSize ||
         !Json::Reader{}.parse(message, jv) || !jv.isObject())