ZMQ Hardening (#96)

src/rpc/admin.cpp

@@ -49,7 +49,7 @@ namespace wire
 {
   static void write_bytes(wire::writer& dest, const std::pair<lws::db::webhook_key, std::vector<lws::db::webhook_value>>& self)
   {
-    wire::object(dest, wire::field<0>("key", self.first), wire::field<1>("value", self.second));
+    wire::object(dest, wire::field<0>("key", std::cref(self.first)), wire::field<1>("value", std::cref(self.second)));
   }
 }
 
@@ -119,8 +119,14 @@ namespace
   template<typename T, typename... U>
   void read_addresses(wire::reader& source, T& self, U... field)
   {
+    using min_address_size =
+      wire::min_element_sizeof<crypto::public_key, crypto::public_key>;
+
     std::vector<std::string> addresses;
-    wire::object(source, wire::field("addresses", std::ref(addresses)), std::move(field)...);
+    wire::object(source,
+      wire::field("addresses", wire::array<min_address_size>(std::ref(addresses))),
+      std::move(field)...
+    );
 
     self.addresses.reserve(addresses.size());
     for (const auto& elem : addresses)

src/rpc/client.cpp

@@ -58,6 +58,8 @@ namespace rpc
     constexpr const char minimal_chain_topic[] = "json-minimal-chain_main";
     constexpr const char full_txpool_topic[] = "json-full-txpool_add";
     constexpr const int daemon_zmq_linger = 0;
+    constexpr const std::int64_t max_msg_sub = 10 * 1024 * 1024;  // 50 MiB
+    constexpr const std::int64_t max_msg_req = 350 * 1024 * 1024; // 350 MiB
     constexpr const std::chrono::seconds chain_poll_timeout{20};
     constexpr const std::chrono::minutes chain_sub_timeout{4};
 
@@ -166,13 +168,20 @@ namespace rpc
       MONERO_ZMQ_CHECK(zmq_setsockopt(signal_sub, ZMQ_SUBSCRIBE, signal, sizeof(signal) - 1));
       return success();
     }
+
+    template<typename T>
+    expect<void> do_set_option(void* sock, const int option, const T value) noexcept
+    {
+      MONERO_ZMQ_CHECK(zmq_setsockopt(sock, option, std::addressof(value), sizeof(value)));
+      return success();
+    }
   } // anonymous
 
   namespace detail
   {
     struct context
     {
-      explicit context(zcontext comm, socket signal_pub, socket external_pub, rcontext rmq, std::string daemon_addr, std::string sub_addr, std::chrono::minutes interval)
+      explicit context(zcontext comm, socket signal_pub, socket external_pub, rcontext rmq, std::string daemon_addr, std::string sub_addr, std::chrono::minutes interval, bool untrusted_daemon)
         : comm(std::move(comm))
         , signal_pub(std::move(signal_pub))
         , external_pub(std::move(external_pub))
@@ -185,6 +194,7 @@ namespace rpc
         , cached{}
         , sync_pub()
         , sync_rates()
+        , untrusted_daemon(untrusted_daemon)
       {
         if (std::chrono::minutes{0} < cache_interval)
           rates_conn.set_server(crypto_compare.host, boost::none, epee::net_utils::ssl_support_t::e_ssl_support_enabled);
@@ -202,6 +212,7 @@ namespace rpc
       rates cached;
       boost::mutex sync_pub;
       boost::mutex sync_rates;
+      const bool untrusted_daemon;
     };
   } // detail
 
@@ -254,14 +265,15 @@ namespace rpc
   {
     MONERO_PRECOND(ctx != nullptr);
 
-    int option = daemon_zmq_linger;
     client out{std::move(ctx)};
 
     out.daemon.reset(zmq_socket(out.ctx->comm.get(), ZMQ_REQ));
     if (out.daemon.get() == nullptr)
       return net::zmq::get_error_code();
+    MONERO_CHECK(do_set_option(out.daemon.get(), ZMQ_LINGER, daemon_zmq_linger));
+    if (out.ctx->untrusted_daemon)
+      MONERO_CHECK(do_set_option(out.daemon.get(), ZMQ_MAXMSGSIZE, max_msg_req));
     MONERO_ZMQ_CHECK(zmq_connect(out.daemon.get(), out.ctx->daemon_addr.c_str()));
-    MONERO_ZMQ_CHECK(zmq_setsockopt(out.daemon.get(), ZMQ_LINGER, &option, sizeof(option)));
 
     if (!out.ctx->sub_addr.empty())
     {
@@ -269,6 +281,8 @@ namespace rpc
       if (out.daemon_sub.get() == nullptr)
         return net::zmq::get_error_code();
 
+      if (out.ctx->untrusted_daemon)
+        MONERO_CHECK(do_set_option(out.daemon_sub.get(), ZMQ_MAXMSGSIZE, max_msg_sub));
       MONERO_ZMQ_CHECK(zmq_connect(out.daemon_sub.get(), out.ctx->sub_addr.c_str()));
       MONERO_CHECK(do_subscribe(out.daemon_sub.get(), minimal_chain_topic));
       MONERO_CHECK(do_subscribe(out.daemon_sub.get(), full_txpool_topic));
@@ -424,7 +438,7 @@ namespace rpc
     return ctx->cached;
   }
 
-  context context::make(std::string daemon_addr, std::string sub_addr, std::string pub_addr, rmq_details rmq_info, std::chrono::minutes rates_interval)
+  context context::make(std::string daemon_addr, std::string sub_addr, std::string pub_addr, rmq_details rmq_info, std::chrono::minutes rates_interval, const bool untrusted_daemon)
   {
     zcontext comm{zmq_init(1)};
     if (comm == nullptr)
@@ -502,7 +516,7 @@ namespace rpc
 
     return context{
       std::make_shared<detail::context>(
-        std::move(comm), std::move(pub), std::move(external_pub), std::move(rmq), std::move(daemon_addr), std::move(sub_addr), rates_interval
+        std::move(comm), std::move(pub), std::move(external_pub), std::move(rmq), std::move(daemon_addr), std::move(sub_addr), rates_interval, untrusted_daemon
       )
     };
   }

src/rpc/client.h

@@ -204,8 +204,10 @@ namespace rpc
       \param rmq_info Required information for RMQ publishing (if enabled)
       \param rates_interval Frequency to retrieve exchange rates. Set value to
         `<= 0` to disable exchange rate retrieval.
+      \param True if additional size constraints should be placed on
+        daemon messages
     */
-    static context make(std::string daemon_addr, std::string sub_addr, std::string pub_addr, rmq_details rmq_info, std::chrono::minutes rates_interval);
+    static context make(std::string daemon_addr, std::string sub_addr, std::string pub_addr, rmq_details rmq_info, std::chrono::minutes rates_interval, const bool untrusted_daemon);
 
     context(context&&) = default;
     context(context const&) = delete;

src/rpc/daemon_pub.cpp

@@ -34,19 +34,24 @@
 #include "wire/field.h"
 #include "wire/traits.h"
 #include "wire/json/read.h"
+#include "wire/wrapper/array.h"
+#include "wire/wrappers_impl.h"
 
 namespace
 {
+  using max_txes_pub = wire::max_element_count<775>;
+
   struct dummy_chain_array
   {
     using value_type = crypto::hash;
 
-    std::uint64_t count;
+    std::size_t count = 0;
     std::reference_wrapper<crypto::hash> id;
 
     void clear() noexcept {}
     void reserve(std::size_t) noexcept {}
 
+    std::size_t size() const noexcept { return count; }
     crypto::hash& back() noexcept { return id; }
     void emplace_back() { ++count; }
   };
@@ -88,7 +93,7 @@ namespace rpc
 
   static void read_bytes(wire::json_reader& source, full_txpool_pub& self)
   {
-    wire_read::array(source, self.txes);
+    wire_read::bytes(source, wire::array<max_txes_pub>(std::ref(self.txes)));
   }
 
   expect<full_txpool_pub> full_txpool_pub::from_json(std::string&& source)

src/rpc/daemon_zmq.cpp

@@ -28,11 +28,14 @@
 #include "daemon_zmq.h"
 
 #include <boost/optional/optional.hpp>
+#include "cryptonote_config.h"        // monero/src
 #include "crypto/crypto.h"            // monero/src
 #include "rpc/message_data_structs.h" // monero/src
 #include "wire/crypto.h"
 #include "wire/json.h"
+#include "wire/wrapper/array.h"
 #include "wire/wrapper/variant.h"
+#include "wire/wrappers_impl.h"
 #include "wire/vector.h"
 
 namespace
@@ -43,6 +46,17 @@ namespace
   constexpr const std::size_t default_outputs = 4;
   constexpr const std::size_t default_txextra_size = 2048;
   constexpr const std::size_t default_txpool_size = 32;
+
+  using max_blocks_per_fetch =
+    wire::max_element_count<COMMAND_RPC_GET_BLOCKS_FAST_MAX_BLOCK_COUNT>;
+
+  //! Not the default in cryptonote, but roughly a 31.8 MiB block
+  using max_txes_per_block = wire::max_element_count<21845>; 
+
+  using max_inputs_per_tx = wire::max_element_count<3000>;
+  using max_outputs_per_tx = wire::max_element_count<2000>;
+  using max_ring_size = wire::max_element_count<4600>;
+  using max_txpool_size = wire::max_element_count<775>;
 }
 
 namespace rct
@@ -65,7 +79,11 @@ namespace rct
 
   static void read_bytes(wire::json_reader& source, mgSig& self)
   {
-    wire::object(source, WIRE_FIELD(ss), WIRE_FIELD(cc));
+    using max_256 = wire::max_element_count<256>;
+    wire::object(source,
+      wire::field("ss", wire::array<max_256>(std::ref(self.ss))),
+      WIRE_FIELD(cc)
+    );
   }
 
   static void read_bytes(wire::json_reader& source, BulletproofPlus& self)
@@ -142,13 +160,20 @@ namespace rct
 
     void read_bytes(wire::json_reader& source, prunable_helper& self)
     {
+      using rf_min_size = wire::min_element_sizeof<key64, key64, key64, key>;
+      using bf_max = wire::max_element_count<BULLETPROOF_MAX_OUTPUTS>;
+      using bf_plus_max = wire::max_element_count<BULLETPROOF_PLUS_MAX_OUTPUTS>;
+      using mlsags_max = wire::max_element_count<256>;
+      using clsags_max = wire::max_element_count<256>;
+      using pseudo_outs_max = wire::max_element_count<256>;
+
       wire::object(source,
-        wire::field("range_proofs", std::ref(self.prunable.rangeSigs)),
-        wire::field("bulletproofs", std::ref(self.prunable.bulletproofs)),
-        wire::field("bulletproofs_plus", std::ref(self.prunable.bulletproofs_plus)),
-        wire::field("mlsags", std::ref(self.prunable.MGs)),
-        wire::field("clsags", std::ref(self.prunable.CLSAGs)),
-        wire::field("pseudo_outs", std::ref(self.pseudo_outs))
+        wire::field("range_proofs", wire::array<rf_min_size>(std::ref(self.prunable.rangeSigs))),
+        wire::field("bulletproofs", wire::array<bf_max>(std::ref(self.prunable.bulletproofs))),
+        wire::field("bulletproofs_plus", wire::array<bf_plus_max>(std::ref(self.prunable.bulletproofs_plus))),
+        wire::field("mlsags", wire::array<mlsags_max>(std::ref(self.prunable.MGs))),
+        wire::field("clsags", wire::array<clsags_max>(std::ref(self.prunable.CLSAGs))),
+        wire::field("pseudo_outs", wire::array<pseudo_outs_max>(std::ref(self.pseudo_outs)))
       );
 
       const bool pruned =
@@ -166,15 +191,16 @@ namespace rct
 
   static void read_bytes(wire::json_reader& source, rctSig& self)
   {
-    boost::optional<std::vector<ecdhTuple>> ecdhInfo;
-    boost::optional<ctkeyV> outPk;
+    using min_ecdh = wire::min_element_sizeof<rct::key, rct::key>;
+    using min_ctkey = wire::min_element_sizeof<rct::key>;
+ 
     boost::optional<xmr_amount> txnFee;
     boost::optional<prunable_helper> prunable;
     self.outPk.reserve(default_inputs);
     wire::object(source,
       WIRE_FIELD(type),
-      wire::optional_field("encrypted", std::ref(ecdhInfo)),
-      wire::optional_field("commitments", std::ref(outPk)),
+      wire::optional_field("encrypted", wire::array<min_ecdh>(std::ref(self.ecdhInfo))),
+      wire::optional_field("commitments", wire::array<min_ctkey>(std::ref(self.outPk))),
       wire::optional_field("fee", std::ref(txnFee)),
       wire::optional_field("prunable", std::ref(prunable))
     );
@@ -182,13 +208,11 @@ namespace rct
     self.txnFee = 0;
     if (self.type != RCTTypeNull)
     {
-      if (!ecdhInfo || !outPk || !txnFee)
+      if (self.ecdhInfo.empty() || self.outPk.empty() || !txnFee)
         WIRE_DLOG_THROW(wire::error::schema::missing_key, "Expected fields `encrypted`, `commitments`, and `fee`");
-      self.ecdhInfo = std::move(*ecdhInfo);
-      self.outPk = std::move(*outPk);
       self.txnFee = std::move(*txnFee);
     }
-    else if (ecdhInfo || outPk || txnFee)
+    else if (!self.ecdhInfo.empty() || !self.outPk.empty() || txnFee)
       WIRE_DLOG_THROW(wire::error::schema::invalid_key, "Did not expected `encrypted`, `commitments`, or `fee`");
 
     if (prunable)
@@ -243,7 +267,11 @@ namespace cryptonote
   }
   static void read_bytes(wire::json_reader& source, txin_to_key& self)
   {
-    wire::object(source, WIRE_FIELD(amount), WIRE_FIELD(key_offsets), wire::field("key_image", std::ref(self.k_image)));
+    wire::object(source,
+      WIRE_FIELD(amount),
+      WIRE_FIELD_ARRAY(key_offsets, max_ring_size),
+      wire::field("key_image", std::ref(self.k_image))
+    );
   }
   static void read_bytes(wire::json_reader& source, txin_v& self)
   {
@@ -264,35 +292,47 @@ namespace cryptonote
     wire::object(source,
       WIRE_FIELD(version),
       WIRE_FIELD(unlock_time),
-      wire::field("inputs", std::ref(self.vin)),
-      wire::field("outputs", std::ref(self.vout)),
+      wire::field("inputs", wire::array<max_inputs_per_tx>(std::ref(self.vin))),
+      wire::field("outputs", wire::array<max_outputs_per_tx>(std::ref(self.vout))),
       WIRE_FIELD(extra),
+      WIRE_FIELD_ARRAY(signatures, max_inputs_per_tx),
       wire::field("ringct", std::ref(self.rct_signatures))
     );
   }
 
   static void read_bytes(wire::json_reader& source, block& self)
   {
+    using min_hash_size = wire::min_element_sizeof<crypto::hash>;
     self.tx_hashes.reserve(default_transaction_count);
     wire::object(source,
       WIRE_FIELD(major_version),
       WIRE_FIELD(minor_version),
       WIRE_FIELD(timestamp),
       WIRE_FIELD(miner_tx),
-      WIRE_FIELD(tx_hashes),
+      WIRE_FIELD_ARRAY(tx_hashes, min_hash_size),
       WIRE_FIELD(prev_id),
       WIRE_FIELD(nonce)
     );
   }
 
-  namespace rpc
+  static void read_bytes(wire::json_reader& source, std::vector<transaction>& self)
   {
+    wire_read::array_unchecked(source, self, 0, max_txes_per_block{});
+  }
+
+  namespace rpc
+  { 
     static void read_bytes(wire::json_reader& source, block_with_transactions& self)
     {
       self.transactions.reserve(default_transaction_count);
       wire::object(source, WIRE_FIELD(block), WIRE_FIELD(transactions));
     }
 
+    static void read_bytes(wire::json_reader& source, std::vector<block_with_transactions>& self)
+    {
+      wire_read::array_unchecked(source, self, 0, max_blocks_per_fetch{}); 
+    }
+
     static void read_bytes(wire::json_reader& source, tx_in_pool& self)
     {
       wire::object(source, WIRE_FIELD(tx), WIRE_FIELD(tx_hash));
@@ -310,11 +350,16 @@ void lws::rpc::read_bytes(wire::json_reader& source, get_blocks_fast_response& s
 {
   self.blocks.reserve(default_blocks_fetched);
   self.output_indices.reserve(default_blocks_fetched);
-  wire::object(source, WIRE_FIELD(blocks), WIRE_FIELD(output_indices), WIRE_FIELD(start_height), WIRE_FIELD(current_height));
+  wire::object(source,
+    WIRE_FIELD(blocks),
+    wire::field("output_indices", wire::array<max_blocks_per_fetch>(wire::array<max_txes_per_block>(wire::array<max_outputs_per_tx>(std::ref(self.output_indices))))),
+    WIRE_FIELD(start_height),
+    WIRE_FIELD(current_height)
+  );
 }
 
 void lws::rpc::read_bytes(wire::json_reader& source, get_transaction_pool_response& self)
 {
   self.transactions.reserve(default_txpool_size);
-  wire::object(source, WIRE_FIELD(transactions));
+  wire::object(source, WIRE_FIELD_ARRAY(transactions, max_txpool_size));
 }

src/rpc/light_wallet.cpp

@@ -50,6 +50,8 @@
 
 namespace
 {
+  using max_subaddrs = wire::max_element_count<16384>;
+
   enum class iso_timestamp : std::uint64_t {};
 
   struct rct_bytes
@@ -178,7 +180,7 @@ namespace lws
   }
   void rpc::read_bytes(wire::json_reader& source, safe_uint64_array& self)
   {
-    for (std::size_t count = source.start_array(); !source.is_array_end(count); --count)
+    for (std::size_t count = source.start_array(0); !source.is_array_end(count); --count)
       self.values.emplace_back(wire::integer::cast_unsigned<std::uint64_t>(source.safe_unsigned_integer()));
     source.end_array();
   }
@@ -374,7 +376,7 @@ namespace lws
     wire::object(source,
       wire::field("address", std::ref(address)),
       wire::field("view_key", std::ref(unwrap(unwrap(self.creds.key)))),
-      WIRE_FIELD(subaddrs),
+      WIRE_FIELD_ARRAY(subaddrs, max_subaddrs),
       WIRE_OPTIONAL_FIELD(get_all)
     );
     convert_address(address, self.creds.address);