fix: Set request size limits and differential pricing for get-object-by-hash calls

src/test/overlay/TMGetObjectByHash_test.cpp

@@ -100,6 +100,17 @@ class TMGetObjectByHash_test : public beast::unit_test::Suite
             return lastSentMessage_;
         }
 
+        // Synchronous test access to the JobQueue-dispatched processor.
+        // The production path runs this on JtLedgerReq; tests need a
+        // synchronous entry point to inspect the reply via send().
+        // PeerImp::processGetObjectByHash is `protected` so the derived
+        // test subclass can call it directly.
+        void
+        runProcessGetObjectByHash(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
+        {
+            processGetObjectByHash(m);
+        }
+
         static void
         resetId()
         {
@@ -179,6 +190,10 @@ class TMGetObjectByHash_test : public beast::unit_test::Suite
     /**
      * Test that reply is limited to hardMaxReplyNodes when more objects
      * are requested than the limit allows.
+     *
+     * `onMessage(TMGetObjectByHash)` dispatches the generic-query path
+     * to the JobQueue, so tests invoke the synchronous processor
+     * directly via `runProcessGetObjectByHash`.
      */
     void
     testReplyLimit(size_t const numObjects, int const expectedReplySize)
@@ -191,8 +206,7 @@ class TMGetObjectByHash_test : public beast::unit_test::Suite
         auto peer = createPeer(env);
 
         auto request = createRequest(numObjects, env);
-        // Call the onMessage handler
-        peer->onMessage(request);
+        peer->runProcessGetObjectByHash(request);
 
         // Verify that a reply was sent
         auto sentMessage = peer->getLastSentMessage();

src/xrpld/overlay/detail/PeerImp.cpp

@@ -22,6 +22,7 @@
 #include <xrpld/peerfinder/PeerfinderManager.h>
 #include <xrpld/peerfinder/Slot.h>
 
+#include <xrpl/basics/Blob.h>
 #include <xrpl/basics/Log.h>
 #include <xrpl/basics/SHAMapHash.h>
 #include <xrpl/basics/Slice.h>
@@ -81,6 +82,7 @@
 #include <xrpl.pb.h>
 
 #include <algorithm>
+#include <atomic>
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
@@ -393,11 +395,21 @@ void
 PeerImp::charge(Resource::Charge const& fee, std::string const& context)
 {
     if ((usage_.charge(fee, context) == Resource::Disposition::Drop) &&
-        usage_.disconnect(pJournal_) && strand_.running_in_this_thread())
+        usage_.disconnect(pJournal_))
     {
-        // Sever the connection
-        overlay_.incPeerDisconnectCharges();
-        fail("charge: Resources");
+        // Idempotent: only the first worker to observe Drop counts the
+        // metric and posts fail().  Without the guard, several queued
+        // workers can all see Drop before fail() lands on the strand,
+        // overcounting peerDisconnectsCharges_ and posting duplicate
+        // shutdowns.  fail(std::string const&) self-posts to strand_
+        // when invoked off-strand.
+        bool expected = false;
+        if (chargeDisconnectFired_.compare_exchange_strong(
+                expected, true, std::memory_order_acq_rel))
+        {
+            overlay_.incPeerDisconnectCharges();
+            fail("charge: Resources");
+        }
     }
 }
 
@@ -2473,63 +2485,63 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
             return;
         }
 
-        protocol::TMGetObjectByHash reply;
-
-        reply.set_query(false);
-
-        reply.set_type(packet.type());
-
         if (packet.has_ledgerhash())
         {
             if (!stringIsUInt256Sized(packet.ledgerhash()))
             {
-                fee_.update(Resource::kFeeMalformedRequest, "ledger hash");
+                JLOG(pJournal_.debug()) << "GetObj: malformed ledgerhash from peer " << id_;
+                fee_.update(Resource::kFeeMalformedRequest, "get object ledger hash");
                 return;
             }
-
-            reply.set_ledgerhash(packet.ledgerhash());
         }
-
-        fee_.update(Resource::kFeeModerateBurdenPeer, " received a get object by hash request");
-
-        // This is a very minimal implementation
-        for (int i = 0; i < packet.objects_size(); ++i)
+        // Reject oversized requests before touching the NodeStore.
+        // The legitimate upper bound (InboundLedger::getNeededHashes())
+        // is 8 hashes; anything beyond kHardMaxReplyNodes is non-conforming.
+        if (packet.objects_size() > Tuning::kHardMaxReplyNodes)
         {
-            auto const& obj = packet.objects(i);
-            if (obj.has_hash() && stringIsUInt256Sized(obj.hash()))
-            {
-                uint256 const hash = uint256::fromRaw(obj.hash());
-                // VFALCO TODO Move this someplace more sensible so we dont
-                //             need to inject the NodeStore interfaces.
-                std::uint32_t const seq{obj.has_ledgerseq() ? obj.ledgerseq() : 0};
-                auto nodeObject{app_.getNodeStore().fetchNodeObject(hash, seq)};
-                if (nodeObject)
-                {
-                    protocol::TMIndexedObject& newObj = *reply.add_objects();
-                    newObj.set_hash(hash.begin(), hash.size());
-                    newObj.set_data(&nodeObject->getData().front(), nodeObject->getData().size());
-
-                    if (obj.has_nodeid())
-                        newObj.set_index(obj.nodeid());
-                    if (obj.has_ledgerseq())
-                        newObj.set_ledgerseq(obj.ledgerseq());
-
-                    // Check if by adding this object, reply has reached its
-                    // limit
-                    if (reply.objects_size() >= Tuning::kHardMaxReplyNodes)
-                    {
-                        fee_.update(
-                            Resource::kFeeModerateBurdenPeer,
-                            "Reply limit reached. Truncating reply.");
-                        break;
-                    }
-                }
-            }
+            JLOG(pJournal_.warn())
+                << "GetObj: oversized request from peer " << id_ << " (" << packet.objects_size()
+                << " > " << Tuning::kHardMaxReplyNodes << ")";
+            fee_.update(Resource::kFeeInvalidData, "oversized get object request");
+            return;
         }
 
-        JLOG(pJournal_.trace()) << "GetObj: " << reply.objects_size() << " of "
-                                << packet.objects_size();
-        send(std::make_shared<Message>(reply, protocol::mtGET_OBJECTS));
+        // Dispatch heavy synchronous NodeStore lookups off the peer's
+        // I/O strand and onto the bounded job queue, mirroring the pattern
+        // used by processLedgerRequest.
+        std::weak_ptr<PeerImp> const weak = shared_from_this();
+        bool const queued = app_.getJobQueue().addJob(JtLedgerReq, "RcvGetObjByHash", [weak, m]() {
+            auto peer = weak.lock();
+            if (!peer)
+                return;
+            try
+            {
+                peer->processGetObjectByHash(m);
+            }
+            catch (std::exception const& e)
+            {
+                // Surface backend failures (NodeStore I/O, allocation)
+                // back through the resource model so a misbehaving peer
+                // is still accountable rather than silently dropped.
+                JLOG(peer->pJournal_.warn()) << "GetObj: handler threw: " << e.what();
+                peer->charge(Resource::kFeeRequestNoReply, "get object handler exception");
+            }
+        });
+        if (!queued)
+        {
+            // The JobQueue is no longer accepting new work (typically
+            // because it is shutting down / has been joined).
+            JLOG(pJournal_.warn()) << "GetObj: job queue refused request from peer " << id_;
+            return;
+        }
+
+        // Admission-time charge: a peer that floods enqueues would
+        // otherwise be billed only the trivial onMessageEnd fee per
+        // message until the JobQueue catches up, re-creating an
+        // uncharged DoS window. Charge the base burden up-front (after
+        // a successful enqueue); the per-lookup differential is added
+        // in the worker.
+        fee_.update(Resource::kFeeModerateBurdenPeer, "received a get object by hash request");
     }
     else
     {
@@ -2585,6 +2597,69 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
     }
 }
 
+void
+PeerImp::processGetObjectByHash(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
+{
+    protocol::TMGetObjectByHash const& packet = *m;
+
+    protocol::TMGetObjectByHash reply;
+    reply.set_query(false);
+    reply.set_type(packet.type());
+
+    if (packet.has_ledgerhash())
+    {
+        reply.set_ledgerhash(packet.ledgerhash());
+    }
+
+    // Defense in depth: caller (onMessage) already validates cheap
+    // structural properties of the request before dispatching here:
+    //   - objects_size() <= kHardMaxReplyNodes (oversize gate)
+    //   - if has_ledgerhash() then ledgerhash is uint256-sized
+    // The iteration cap below mirrors the oversize gate so this method
+    // remains safe if invoked directly by tests or future callers, and
+    // a peer cannot drive unbounded NodeStore lookups by sending
+    // non-existent hashes.
+    int const requested = packet.objects_size();
+    int const iterLimit = std::min(requested, Tuning::kHardMaxReplyNodes);
+
+    for (int i = 0; i < iterLimit; ++i)
+    {
+        auto const& obj = packet.objects(i);
+        if (!obj.has_hash() || !stringIsUInt256Sized(obj.hash()))
+            continue;
+
+        uint256 const hash = uint256::fromRaw(obj.hash());
+        // VFALCO TODO Move this someplace more sensible so we don't
+        //             need to inject the NodeStore interfaces.
+        std::uint32_t const seq{obj.has_ledgerseq() ? obj.ledgerseq() : 0};
+        auto const nodeObject = app_.getNodeStore().fetchNodeObject(hash, seq);
+        if (!nodeObject)
+            continue;
+
+        protocol::TMIndexedObject& newObj = *reply.add_objects();
+        newObj.set_hash(hash.begin(), hash.size());
+        auto const& data = nodeObject->getData();
+        newObj.set_data(data.data(), data.size());
+        if (obj.has_nodeid())
+            newObj.set_index(obj.nodeid());
+        if (obj.has_ledgerseq())
+            newObj.set_ledgerseq(obj.ledgerseq());
+    }
+
+    // Apply work-proportional charge. `charge()` posts the disconnect
+    // step (if any) back to strand_, so it is safe to call from this
+    // JobQueue worker thread.
+    charge(
+        // We pass `requested` directly here, instead of actual lookups done. Which could be
+        // std::min(packet.objects_size(), static_cast<int>(Tuning::kHardMaxReplyNodes));
+        // Because we want to charge as per the request size, to discourage large requests.
+        computeGetObjectByHashFee(requested, reply.objects_size()),
+        "processed get object by hash request");
+
+    JLOG(pJournal_.trace()) << "GetObj: " << reply.objects_size() << " of " << requested;
+    send(std::make_shared<Message>(reply, protocol::mtGET_OBJECTS));
+}
+
 void
 PeerImp::onMessage(std::shared_ptr<protocol::TMHaveTransactions> const& m)
 {
@@ -3412,6 +3487,53 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m)
     send(std::make_shared<Message>(ledgerData, protocol::mtLEDGER_DATA));
 }
 
+// Differential pricing helper.  Returns only the *dynamic* component
+// of the per-message charge — the base `kFeeModerateBurdenPeer` is
+// applied at admission time in `onMessage(TMGetObjectByHash)` so a
+// high traffic client pays for the message regardless of when (or
+// whether) the worker runs.
+//
+// Dynamic charge model:
+//
+//   billable       = max(0, requested - kFreeObjectsPerRequest)
+//   missed         = max(0, requested - found)
+//   billableMisses = min(missed, billable)        // misses billed first
+//   billableHits   = billable - billableMisses
+//   sizeBand       = (requested >  kBandMediumMax) ? kCostBandLarge
+//                  : (requested >  kBandSmallMax)  ? kCostBandMedium
+//                                                  : kCostBandSmall
+//   dynamic        = billableHits   * kCostPerLookupHit
+//                  + billableMisses * kCostPerLookupMiss
+//                  + sizeBand
+//
+// Misses are billed first against the billable budget because a node store
+// seek dominates a cache hit and because invalid hashes are ~100% miss by construction.
+Resource::Charge
+PeerImp::computeGetObjectByHashFee(int const requested, int const found)
+{
+    int const billable = std::max(0, requested - static_cast<int>(Tuning::kFreeObjectsPerRequest));
+    // Clamp `missed` so a future caller passing found > requested cannot
+    // produce a negative value that flips the hits/misses split.
+    int const missed = std::max(0, requested - found);
+    int const billableMisses = std::min(missed, billable);
+    int const billableHits = billable - billableMisses;
+
+    int sizeBand = Tuning::kCostBandSmall;
+    if (requested > Tuning::kBandMediumMax)
+    {
+        sizeBand = Tuning::kCostBandLarge;
+    }
+    else if (requested > Tuning::kBandSmallMax)
+    {
+        sizeBand = Tuning::kCostBandMedium;
+    }
+
+    int const dynamic = (billableHits * Tuning::kCostPerLookupHit) +
+        (billableMisses * Tuning::kCostPerLookupMiss) + sizeBand;
+
+    return Resource::Charge(dynamic, "GetObject differential");
+}
+
 int
 PeerImp::getScore(bool haveItem) const
 {

src/xrpld/overlay/detail/PeerImp.h

@@ -147,6 +147,12 @@ private:
     protocol::TMStatusChange lastStatus_;
     Resource::Consumer usage_;
     ChargeWithContext fee_;
+
+    // One-shot guard so concurrent JobQueue workers cannot double-count
+    // the per-connection peer-disconnect-by-charge metric (and cannot
+    // post duplicate fail() calls) when several queued requests cross
+    // kDropThreshold before the first fail() lands on the strand.
+    std::atomic<bool> chargeDisconnectFired_{false};
     std::shared_ptr<PeerFinder::Slot> const slot_;
     boost::beast::multi_buffer readBuffer_;
     http_request_type request_;
@@ -624,6 +630,67 @@ private:
 
     void
     processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m);
+
+protected:
+    // Kept `protected` so test subclasses (see
+    // TMGetObjectByHash_test) can drive the
+    // synchronous processor and the differential-pricing helper without
+    // routing through the JobQueue or going through `friend` plumbing.
+    // Production callers reach these members only via
+    // `onMessage(TMGetObjectByHash)` → JobQueue → `processGetObjectByHash`.
+
+    /** Process a generic-query TMGetObjectByHash message.
+
+        Dispatched from `onMessage(TMGetObjectByHash)` to the JobQueue
+        (`JtLedgerReq`) so synchronous NodeStore lookups do not block the
+        peer's I/O strand. Caps iteration at `Tuning::kHardMaxReplyNodes`
+        regardless of hit/miss outcome and applies differential pricing
+        via `computeGetObjectByHashFee()` after the fetch loop completes.
+
+        @param m The protocol message containing requested object hashes.
+     */
+    void
+    processGetObjectByHash(std::shared_ptr<protocol::TMGetObjectByHash> const& m);
+
+    /** Compute the per-message resource charge for a TMGetObjectByHash
+        request based on how much work was actually performed.
+
+        The charge has three components on top of the base
+        `Resource::kFeeModerateBurdenPeer`:
+          - per-hit lookup cost (cheap; usually served from cache)
+          - per-miss lookup cost (expensive node store seeks)
+          - request-size band surcharge (escalates abusive batch sizes)
+
+        The first `Tuning::kFreeObjectsPerRequest` objects are free so
+        that legitimate `InboundLedger::getNeededHashes()` traffic
+        (at most 8 objects) is unaffected.
+
+        @param requested Number of objects requested by the message. This
+                          value is used for request-size pricing and may
+                          exceed `Tuning::kHardMaxReplyNodes` when this
+                          helper is called directly, even though processing
+                          caps the iterations to `Tuning::kHardMaxReplyNodes`.
+        @param found     Number of objects successfully returned in the
+                         reply.
+        @return A `Resource::Charge` whose cost reflects the work performed.
+     */
+    static Resource::Charge
+    computeGetObjectByHashFee(int const requested, int const found);
+
+    /** Read-only accessor for the accumulated peer-message charge.
+
+        Exposed at `protected` scope so test subclasses can verify the
+        oversized-request rejection path (Layer 1) without invoking the
+        full JobQueue handler. Production callers should never read this back —
+        the value is consumed by `charge()`/`disconnect()` internally.
+
+        @return The current `Resource::Charge` accumulated on `fee_`.
+     */
+    Resource::Charge
+    currentFeeCharge() const
+    {
+        return fee_.fee;
+    }
 };
 
 //------------------------------------------------------------------------------

src/xrpld/overlay/detail/Tuning.h

@@ -1,4 +1,6 @@
 #pragma once
+#include <xrpl/shamap/SHAMapInnerNode.h>
+
 #include <cstddef>
 #include <cstdint>
 
@@ -39,4 +41,92 @@ static constexpr auto kMaxQueryDepth = 3;
 /** Size of buffer used to read from the socket. */
 constexpr std::size_t kReadBufferBytes = 16384;
 
+/** TMGetObjectByHash differential pricing.
+
+    Honest peers ask for at most 8 hashes per call (the header, or up to
+    4 state + 4 tx hashes from `InboundLedger::getNeededHashes()`). The
+    free tier covers them at zero cost. Beyond that, each lookup is billed:
+    'misses' cost much more than 'hits' because a miss does a node store seek
+    while a hit is usually served from cache. On top of that, a size-band
+    surcharge kicks in for larger requests so an attacker who crams a
+    single message with thousands of hashes blows past
+    `Resource::kDropThreshold` and gets disconnected.
+
+    The numbers below are picked to keep three things true given
+    `kDropThreshold = 25000`:
+
+      - Honest traffic (<= 8 objects per request) is free.
+      - A single all-miss request at `kHardMaxReplyNodes` (12288) costs
+        more than the drop threshold, so an attacker gets dropped in one
+        message.
+      - A peer spamming 1024-object hit-only requests gets dropped in
+        ~19 messages — fast enough to be useful, slow enough that an
+        honest peer momentarily sending oversized requests has time to
+        back off. */
+
+/** How many objects a request can ask for before per-lookup billing
+    begins?
+    Twice the honest peak (8) so a peer that occasionally retries a hash
+    never trips pricing. Same value as `SHAMapInnerNode::kBranchFactor`;
+    that's a coincidence, not a requirement. */
+static constexpr auto kFreeObjectsPerRequest = 16;
+
+/** Cost of one cache-hit lookup. The unit; everything else is a
+    multiple of this. */
+static constexpr auto kCostPerLookupHit = 1;
+
+/** Cost of one node-store miss, in units of `kCostPerLookupHit`.
+
+    A miss does a node store disk seek; a hit usually comes from cache.
+    The 8x ratio is an order-of-magnitude guess at the latency gap on
+    SSD-backed nodes, not a measured number. The math only requires this
+    to be at least 2 — any smaller and a full-miss request at the hard
+    cap wouldn't trip the drop threshold. 8 leaves headroom: if
+    `kDropThreshold` goes up or `kHardMaxReplyNodes` comes down, the
+    drop-on-attack property still holds without a code change. */
+static constexpr auto kCostPerLookupMiss = 8;
+
+/** Size-band surcharges. Whichever band a request's size falls into,
+    its surcharge is added once on top of the per-lookup cost.
+
+    The job of the surcharge is to make crossing a band edge feel like
+    a step, not a slope. With these values, the cost roughly doubles or triples at each cliff:
+
+        n=64: costs 48 => n=65 costs 149 (~3x jump)
+        n=1024: costs 1108 => n=1025 costs 2009 (~2x jump)
+
+    The 10x step between medium and large mirrors the ~16x step
+    between the band edges (64 -> 1024) so the cliff feels comparable
+    at both scales.
+ */
+static constexpr auto kCostBandSmall = 0;
+static constexpr auto kCostBandMedium = 100;
+static constexpr auto kCostBandLarge = 1000;
+
+/** How many hashes per type an honest peer asks for at a time.
+
+    Matches the `4` passed to `neededStateHashes(4)` and
+    `neededTxHashes(4)` in `InboundLedger::getNeededHashes()`. Kept here
+    instead of imported from the ledger module so overlay stays
+    self-contained; if that `4` ever changes, update this in lockstep or
+    the band thresholds below will start charging honest peers. */
+static constexpr auto kLegitHashesPerType = 4;
+
+/** Cutoffs that decide which size band a request falls into.
+
+    A SHAMap inner node has 16 children; an honest peer asks for 4
+    hashes per type. So:
+
+      kBandSmallMax  = 4 * 16    =   64  // one inner node's worth
+      kBandMediumMax = 4 * 16^2  = 1024  // a depth-2 subtree's worth
+
+    A request up to 64 objects is small (no surcharge); up to 1024 is
+    medium; anything larger is large. The bounds are inclusive: a
+    request of exactly 64 is small, 65 is medium. Anything past 1024 is
+    well beyond what the honest sync path produces, so it's billed at
+    the large rate to drive attack-shaped traffic over the drop
+    threshold quickly. */
+static constexpr auto kBandSmallMax = kLegitHashesPerType * SHAMapInnerNode::kBranchFactor;
+static constexpr auto kBandMediumMax = kBandSmallMax * SHAMapInnerNode::kBranchFactor;
+
 }  // namespace xrpl::Tuning