rpc: recover from oversized ZMQ requests

(cherry picked from commit d1508916b44d08fa18dc8ac74e04e9f92dbfb2c1)

src/net/zmq.cpp

@@ -144,22 +144,27 @@ namespace zmq
                 static constexpr const int max_out = std::numeric_limits<int>::max();
                 const std::string::size_type initial = payload.size();
                 message part{};
+                bool too_large = false;
                 for (;;)
                 {
                     int last = 0;
                     if ((last = zmq_msg_recv(part.handle(), socket, flags)) < 0)
                         return last;
 
-                    if (max_message_size < payload.size() ||
-                        max_message_size - payload.size() < part.size())
-                    {
-                        errno = EMSGSIZE;
-                        return -1;
-                    }
-                    payload.append(part.data(), part.size());
+                    if (!too_large)
+                        too_large =
+                            max_message_size < payload.size() ||
+                            max_message_size - payload.size() < part.size();
+                    if (!too_large)
+                        payload.append(part.data(), part.size());
                     if (!zmq_msg_more(part.handle()))
                         break;
                 }
+                if (too_large)
+                {
+                    errno = EMSGSIZE;
+                    return -1;
+                }
                 const std::string::size_type added = payload.size() - initial;
                 return unsigned(max_out) < added ? max_out : int(added);
             }

src/rpc/message.cpp

@@ -42,6 +42,7 @@ const char* Message::STATUS_RETRY = "Retry";
 const char* Message::STATUS_FAILED = "Failed";
 const char* Message::STATUS_BAD_REQUEST = "Invalid request type";
 const char* Message::STATUS_BAD_JSON = "Malformed json";
+const char* Message::STATUS_REQUEST_TOO_LARGE = "Request too large";
 
 namespace
 {
@@ -245,6 +246,15 @@ epee::byte_slice BAD_JSON(const std::string& error_details)
   return FullMessage::getResponse(fail, invalid);
 }
 
+epee::byte_slice REQUEST_TOO_LARGE()
+{
+  rapidjson::Value invalid;
+  Message fail;
+  fail.status = Message::STATUS_REQUEST_TOO_LARGE;
+  fail.error_details = "Request exceeds maximum message size.";
+  return FullMessage::getResponse(fail, invalid);
+}
+
 
 }  // namespace rpc
 

src/rpc/message.h

@@ -53,6 +53,7 @@ namespace rpc
       static const char* STATUS_FAILED;
       static const char* STATUS_BAD_REQUEST;
       static const char* STATUS_BAD_JSON;
+      static const char* STATUS_REQUEST_TOO_LARGE;
 
       Message() : status(STATUS_OK), rpc_version(0) { }
 
@@ -106,6 +107,8 @@ namespace rpc
 
   epee::byte_slice BAD_JSON(const std::string& error_details);
 
+  epee::byte_slice REQUEST_TOO_LARGE();
+
 
 }  // namespace rpc
 

src/rpc/zmq_server.cpp

@@ -35,6 +35,7 @@
 #include <system_error>
 
 #include "byte_slice.h"
+#include "rpc/message.h"
 #include "rpc/zmq_pub.h"
 
 #undef MONERO_DEFAULT_LOG_CATEGORY
@@ -154,7 +155,15 @@ void ZmqServer::serve()
         state->relay_to_pub(relay.get(), pub.get());
 
       if (sockets[1].revents)
-        state->sub_request(MONERO_UNWRAP(net::zmq::receive(pub.get(), ZMQ_DONTWAIT)));
+      {
+        expect<std::string> message = net::zmq::receive(pub.get(), ZMQ_DONTWAIT);
+        if (message)
+          state->sub_request(*message);
+        else if (message == net::zmq::make_error_code(ETERM))
+          MONERO_THROW(message.error(), "ZMQ-PUB context terminated");
+        else if (message != net::zmq::make_error_code(EAGAIN))
+          MERROR("Read failure on ZMQ-PUB: " << message.error().message());
+      }
 
       if (!pub || sockets[2].revents)
       {
@@ -162,8 +171,15 @@ void ZmqServer::serve()
         if (!message)
         {
           // EAGAIN can occur when using `zmq_poll`, which doesn't inspect for message validity
-          if (message != net::zmq::make_error_code(EAGAIN))
-            MONERO_THROW(message.error(), "Read failure on ZMQ-RPC");
+          if (message == net::zmq::make_error_code(EMSGSIZE))
+          {
+            MERROR("ZMQ-RPC request exceeds maximum message size");
+            MONERO_UNWRAP(net::zmq::send(REQUEST_TOO_LARGE(), rep.get()));
+          }
+          else if (message == net::zmq::make_error_code(ETERM))
+            MONERO_THROW(message.error(), "ZMQ-RPC context terminated");
+          else if (message != net::zmq::make_error_code(EAGAIN))
+            MERROR("Read failure on ZMQ-RPC: " << message.error().message());
         }
         else // no errors
         {