Merge pull request #451 from stellar/get-scp-state-fix

Rate limit `GET_SCP_STATE` messages

Author: marta-lokhova

src/overlay/Peer.cpp

@@ -56,6 +56,10 @@ using namespace soci;
 
 namespace
 {
+// Maximum number of GET_SCP_STATE requests per window per peer to respond to. A
+// window defaults to roughly 1 minute.
+constexpr uint32_t GET_SCP_STATE_MAX_RATE = 10;
+
 // Check the signature(s) in `tx`, adding the result to the signature cache in
 // the process. This function requires that background signature verification
 // is enabled and the current thread is the overlay thread.
@@ -1416,15 +1420,15 @@ Peer::recvDontHave(StellarMessage const& msg)
 }
 
 bool
-Peer::process(QueryInfo& queryInfo)
+Peer::process(QueryInfo& queryInfo, std::optional<uint32_t> maxQueriesPerWindow)
 {
     auto const& cfg = mAppConnector.getConfig();
     std::chrono::seconds const QUERY_WINDOW =
         std::chrono::duration_cast<std::chrono::seconds>(
             mAppConnector.getLedgerManager().getExpectedLedgerCloseTime() *
             cfg.MAX_SLOTS_TO_REMEMBER);
-    uint32_t const QUERIES_PER_WINDOW =
-        QUERY_WINDOW.count() * QUERY_RESPONSE_MULTIPLIER;
+    uint32_t const QUERIES_PER_WINDOW = maxQueriesPerWindow.value_or(
+        QUERY_WINDOW.count() * QUERY_RESPONSE_MULTIPLIER);
     if (mAppConnector.now() - queryInfo.mLastTimeStamp >= QUERY_WINDOW)
     {
         queryInfo.mLastTimeStamp = mAppConnector.now();
@@ -1679,6 +1683,13 @@ Peer::recvGetSCPState(StellarMessage const& msg)
 {
     ZoneScoped;
     releaseAssert(threadIsMain());
+    if (!process(mSCPStateQueryInfo, GET_SCP_STATE_MAX_RATE))
+    {
+        CLOG_DEBUG(Overlay, "Dropping GET_SCP_STATE request from {}",
+                   KeyUtils::toShortString(mPeerID));
+        return;
+    }
+    mSCPStateQueryInfo.mNumQueries++;
     uint32 seq = msg.getSCPLedgerSeq();
     mAppConnector.getHerder().sendSCPStateToPeer(seq, shared_from_this());
 }

src/overlay/Peer.h

@@ -276,6 +276,7 @@ class Peer : public std::enable_shared_from_this<Peer>,
     std::shared_ptr<TxAdverts> mTxAdverts;
     QueryInfo mQSetQueryInfo;
     QueryInfo mTxSetQueryInfo;
+    QueryInfo mSCPStateQueryInfo;
     bool mPeersReceived{false};
 
     static Hash pingIDfromTimePoint(VirtualClock::time_point const& tp);
@@ -319,7 +320,17 @@ class Peer : public std::enable_shared_from_this<Peer>,
     void sendDontHave(MessageType type, uint256 const& itemID);
     void sendPeers();
     void sendError(ErrorCode error, std::string const& message);
-    bool process(QueryInfo& queryInfo);
+
+    // Returns `true` if a query is within the configured limits and should be
+    // processed, `false` if it exceeds limits and should be dropped. This
+    // function may reset the per-window state in `queryInfo` when a new
+    // window begins, but it does not increment any query counters; callers
+    // are responsible for updating `queryInfo` (for example, incrementing the
+    // number of processed queries) after a query is accepted. Callers may
+    // optionally set `maxQueriesPerWindow` to override the default per-window
+    // query limit.
+    bool process(QueryInfo& queryInfo,
+                 std::optional<uint32_t> maxQueriesPerWindow = std::nullopt);
 
     void recvMessage(std::shared_ptr<CapacityTrackedMessage> msgTracker);
 
@@ -492,6 +503,13 @@ class Peer : public std::enable_shared_from_this<Peer>,
     static void
     populateSignatureCacheForTesting(AppConnector& app,
                                      TransactionFrameBaseConstPtr tx);
+
+    uint32_t
+    getSCPStateQueryCountForTesting() const
+    {
+        releaseAssert(threadIsMain());
+        return mSCPStateQueryInfo.mNumQueries;
+    }
 #endif
 
     // Public thread-safe methods that access Peer's state

src/overlay/test/OverlayTests.cpp

@@ -1843,6 +1843,84 @@ TEST_CASE("drop peers who straggle", "[overlay][connections][straggler]")
     }
 }
 
+TEST_CASE("GET_SCP_STATE rate limiting", "[overlay]")
+{
+    VirtualClock clock;
+    Config cfg1 = getTestConfig(0);
+    Config cfg2 = getTestConfig(1);
+
+    // Bump up close time and max slots to remember to production levels. These
+    // must be large enough that crankSome calls between requests don't cause a
+    // window reset.
+    cfg1.ARTIFICIALLY_SET_CLOSE_TIME_FOR_TESTING = 5;
+    cfg2.ARTIFICIALLY_SET_CLOSE_TIME_FOR_TESTING = 5;
+    cfg1.MAX_SLOTS_TO_REMEMBER = 12;
+    cfg2.MAX_SLOTS_TO_REMEMBER = 12;
+
+    // The window size + 1 second. Minimum time required to ensure the rate
+    // limit window resets.
+    std::chrono::seconds const WINDOW_CLEAR_DURATION(
+        cfg1.ARTIFICIALLY_SET_CLOSE_TIME_FOR_TESTING *
+            cfg1.MAX_SLOTS_TO_REMEMBER +
+        1);
+
+    // Should be no more than 10 processed GET_SCP_STATE messages per window
+    uint32_t constexpr MAX_PER_WINDOW = 10;
+
+    auto app1 = createTestApplication(clock, cfg1);
+    auto app2 = createTestApplication(clock, cfg2);
+
+    LoopbackPeerConnection conn(*app1, *app2);
+    auto sender = conn.getInitiator();
+    auto receiver = conn.getAcceptor();
+    testutil::crankSome(clock);
+    REQUIRE(conn.getInitiator()->isAuthenticatedForTesting());
+    REQUIRE(conn.getAcceptor()->isAuthenticatedForTesting());
+
+    // Advance past QUERY_WINDOW so the first test request triggers a window
+    // reset
+    testutil::crankFor(clock, WINDOW_CLEAR_DURATION);
+
+    // Send requests up to the limit.
+    for (int i = 0; i < MAX_PER_WINDOW; i++)
+    {
+        sender->sendGetScpState(0);
+        testutil::crankSome(clock);
+    }
+
+    // Should have logged 10 queries, and the peers should remain connected
+    REQUIRE(receiver->getSCPStateQueryCountForTesting() == MAX_PER_WINDOW);
+
+    // Send 5 more -- all should be dropped
+    for (int i = 0; i < 5; i++)
+    {
+        sender->sendGetScpState(0);
+        testutil::crankSome(clock);
+    }
+    // Should still be at the maximum query count, as the additional messages
+    // should have been dropped
+    REQUIRE(receiver->getSCPStateQueryCountForTesting() == MAX_PER_WINDOW);
+
+    // Advance past the window duration so the next request triggers a window
+    // reset
+    testutil::crankFor(clock, WINDOW_CLEAR_DURATION);
+
+    // Send a GET_SCP_STATE message to trigger the window reset
+    sender->sendGetScpState(0);
+    testutil::crankSome(clock);
+
+    // Should just have processed the one message sent after the window clear
+    // duration
+    REQUIRE(receiver->getSCPStateQueryCountForTesting() == 1);
+
+    // Peers should still be connected
+    REQUIRE(sender->isConnectedForTesting());
+    REQUIRE(receiver->isConnectedForTesting());
+
+    testutil::shutdownWorkScheduler(*app2);
+    testutil::shutdownWorkScheduler(*app1);
+}
+
 TEST_CASE("reject peers with the same nodeid", "[overlay][connections]")
 {
     VirtualClock clock;