From 43a8be7429d991f35ae920a9e8346722ededf5cf Mon Sep 17 00:00:00 2001 From: Nicholas Dudfield Date: Tue, 14 Jul 2026 14:27:47 +0700 Subject: [PATCH] feat(overlay): relay admitted export shares --- include/xrpl/proto/ripple.proto | 10 + .../overlay/ExportShareTransport_test.cpp | 192 ++++++++++++++++++ src/xrpld/overlay/Overlay.h | 30 +++ src/xrpld/overlay/detail/Message.cpp | 1 + src/xrpld/overlay/detail/OverlayImpl.cpp | 93 +++++++++ src/xrpld/overlay/detail/OverlayImpl.h | 15 ++ src/xrpld/overlay/detail/PeerImp.cpp | 51 +++++ src/xrpld/overlay/detail/PeerImp.h | 2 + src/xrpld/overlay/detail/ProtocolMessage.h | 47 +++++ src/xrpld/overlay/detail/TrafficCount.cpp | 3 + src/xrpld/overlay/detail/TrafficCount.h | 4 + 11 files changed, 448 insertions(+) create mode 100644 src/test/overlay/ExportShareTransport_test.cpp diff --git a/include/xrpl/proto/ripple.proto b/include/xrpl/proto/ripple.proto index 8beeb98b7..2a8848533 100644 --- a/include/xrpl/proto/ripple.proto +++ b/include/xrpl/proto/ripple.proto @@ -27,6 +27,7 @@ enum MessageType mtREPLAY_DELTA_RESPONSE = 60; mtHAVE_TRANSACTIONS = 63; mtTRANSACTIONS = 64; + mtEXPORT_SHARES = 65; } // token, iterations, target, challenge = issue demand for proof of work @@ -120,6 +121,15 @@ message TMTransactions repeated TMTransaction transactions = 1; } +// Canonical post-validation Export signature contributions. Receivers enforce +// ExportLimits caps for entry count, each frame, aggregate frame payload, and +// encoded protobuf size. The encoded-size cap includes protobuf tags and +// length prefixes but excludes the overlay message header. +message TMExportShares +{ + repeated bytes shares = 1; +} + enum NodeStatus { diff --git a/src/test/overlay/ExportShareTransport_test.cpp b/src/test/overlay/ExportShareTransport_test.cpp new file mode 100644 index 000000000..e2e346d2c --- /dev/null +++ b/src/test/overlay/ExportShareTransport_test.cpp @@ -0,0 +1,192 @@ +//------------------------------------------------------------------------------ +/* + This file is part of rippled: https://github.com/ripple/rippled + + Permission to use, copy, modify, and/or distribute this software for any + purpose with or without fee is hereby granted, provided that the above + copyright notice and this permission notice appear in all copies. + + THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES + WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF + MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR + ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES WHATSOEVER + RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF + CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN + CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. +*/ +//============================================================================== + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include + +namespace ripple { +namespace test { + +class ExportShareTransport_test : public beast::unit_test::suite +{ + static ExportShare + makeShare() + { + auto const [key, secret] = randomKeyPair(KeyType::secp256k1); + auto const signature = sign(key, secret, Slice{"export-share", 12}); + return ExportShare{ + ExportShare::currentVersion, + calcAccountID(key), + uint256{1}, + 4'200'000, + uint256{2}, + uint256{3}, + 17, + key, + signature}; + } + + struct Handler + { + bool seen = false; + int shares = 0; + + bool + compressionEnabled() const + { + return false; + } + + void + onMessageUnknown(std::uint16_t) + { + } + + void + onMessageBegin( + std::uint16_t, + std::shared_ptr<::google::protobuf::Message> const&, + std::size_t, + std::size_t, + bool) + { + } + + void + onMessageEnd( + std::uint16_t, + std::shared_ptr<::google::protobuf::Message> const&) + { + } + + template + void + onMessage(std::shared_ptr const& message) + { + if constexpr (std::is_same_v) + { + seen = true; + shares = message->shares_size(); + } + } + }; + + void + testBatchBounds() + { + testcase("ExportShare batch structural bounds"); + + auto const encoded = makeShare().serialize(); + protocol::TMExportShares batch; + batch.add_shares(encoded.data(), encoded.size()); + + auto const parsed = detail::parseExportShareBatch(batch); + BEAST_EXPECT(parsed && parsed->size() == 1); + BEAST_EXPECT( + batch.ByteSizeLong() <= + ExportLimits::maxExportShareRelayMessageBytes); + + protocol::TMExportShares maxBatch; + for (std::size_t i = 0; i < ExportLimits::maxExportSharesPerRelay; ++i) + maxBatch.add_shares(encoded.data(), encoded.size()); + auto const payloadBytes = + encoded.size() * ExportLimits::maxExportSharesPerRelay; + BEAST_EXPECT( + payloadBytes <= ExportLimits::maxExportShareRelayPayloadBytes); + BEAST_EXPECT(maxBatch.ByteSizeLong() > payloadBytes); + BEAST_EXPECT( + maxBatch.ByteSizeLong() <= + ExportLimits::maxExportShareRelayMessageBytes); + BEAST_EXPECT(detail::parseExportShareBatch(maxBatch)); + + protocol::TMExportShares empty; + BEAST_EXPECT(!detail::parseExportShareBatch(empty)); + + protocol::TMExportShares tooMany; + for (std::size_t i = 0; i <= ExportLimits::maxExportSharesPerRelay; ++i) + tooMany.add_shares(encoded.data(), encoded.size()); + BEAST_EXPECT(!detail::parseExportShareBatch(tooMany)); + + protocol::TMExportShares oversized; + oversized.add_shares( + std::string(ExportLimits::maxSerializedExportShareBytes + 1, 'x')); + BEAST_EXPECT(!detail::parseExportShareBatch(oversized)); + + protocol::TMExportShares malformed; + malformed.add_shares("not-an-export-share"); + BEAST_EXPECT(!detail::parseExportShareBatch(malformed)); + } + + void + testDispatchAndTraffic() + { + testcase("ExportShare protocol dispatch and traffic category"); + + auto const encoded = makeShare().serialize(); + protocol::TMExportShares batch; + batch.add_shares(encoded.data(), encoded.size()); + + BEAST_EXPECT(protocol::mtEXPORT_SHARES == 65); + BEAST_EXPECT( + protocolMessageName(protocol::mtEXPORT_SHARES) == "export_shares"); + BEAST_EXPECT( + TrafficCount::categorize(batch, protocol::mtEXPORT_SHARES, true) == + TrafficCount::category::export_shares); + + Message message{batch, protocol::mtEXPORT_SHARES}; + auto const& wire = message.getBuffer(compression::Compressed::Off); + boost::beast::multi_buffer buffers; + buffers.commit(boost::asio::buffer_copy( + buffers.prepare(wire.size()), boost::asio::buffer(wire))); + + Handler handler; + std::size_t hint = 0; + auto const [consumed, ec] = + invokeProtocolMessage(buffers.data(), handler, hint); + BEAST_EXPECT(!ec); + BEAST_EXPECT(consumed == wire.size()); + BEAST_EXPECT(handler.seen); + BEAST_EXPECT(handler.shares == 1); + } + +public: + void + run() override + { + testBatchBounds(); + testDispatchAndTraffic(); + } +}; + +BEAST_DEFINE_TESTSUITE(ExportShareTransport, overlay, ripple); + +} // namespace test +} // namespace ripple diff --git a/src/xrpld/overlay/Overlay.h b/src/xrpld/overlay/Overlay.h index 9aff15606..2767cc98a 100644 --- a/src/xrpld/overlay/Overlay.h +++ b/src/xrpld/overlay/Overlay.h @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -64,6 +65,13 @@ protected: public: enum class Promote { automatic, never, always }; + /** Application admission for a structurally valid ExportShare. + + Returning true means the application has verified the share against + its origin context, committee position/key, and target multisignature. + */ + using ExportShareHandler = std::function; + struct Setup { explicit Setup() = default; @@ -149,6 +157,28 @@ public: virtual void broadcast(protocol::TMValidation& m) = 0; + /** Broadcast application-admitted canonical Export shares. */ + virtual void + broadcast(protocol::TMExportShares& m) = 0; + + /** Relay application-admitted Export shares using raw-frame suppression. */ + virtual void + relay(protocol::TMExportShares& m) = 0; + + /** Install the application admission callback for inbound Export shares. + + The callback runs on the peer job queue, not an overlay I/O strand. + */ + virtual void + setExportShareHandler(ExportShareHandler handler) = 0; + + /** Run application admission for an inbound Export share. + + With no callback installed, admission fails closed. + */ + virtual bool + acceptExportShare(ExportShare const& share) = 0; + /** Relay a proposal. * @param m the serialized proposal * @param uid the id used to identify this proposal diff --git a/src/xrpld/overlay/detail/Message.cpp b/src/xrpld/overlay/detail/Message.cpp index cdcf433f6..b42bc3295 100644 --- a/src/xrpld/overlay/detail/Message.cpp +++ b/src/xrpld/overlay/detail/Message.cpp @@ -101,6 +101,7 @@ Message::compress() case protocol::mtPROOF_PATH_RESPONSE: case protocol::mtREPLAY_DELTA_REQ: case protocol::mtHAVE_TRANSACTIONS: + case protocol::mtEXPORT_SHARES: break; } return false; diff --git a/src/xrpld/overlay/detail/OverlayImpl.cpp b/src/xrpld/overlay/detail/OverlayImpl.cpp index f153f0df0..cf99067ce 100644 --- a/src/xrpld/overlay/detail/OverlayImpl.cpp +++ b/src/xrpld/overlay/detail/OverlayImpl.cpp @@ -1161,6 +1161,99 @@ OverlayImpl::broadcast(protocol::TMValidation& m) for_each([sm](std::shared_ptr&& p) { p->send(sm); }); } +void +OverlayImpl::broadcast(protocol::TMExportShares& m) +{ + auto const shares = detail::parseExportShareBatch(m); + if (!shares) + { + JLOG(journal_.error()) + << "Refusing to broadcast malformed ExportShare batch"; + return; + } + + // Local callers are responsible for application admission before using + // this API. Register raw-wire suppression before routing the batch. + for (auto const& share : *shares) + app_.getHashRouter().addSuppression(share.wireHash()); + + relay(m); +} + +void +OverlayImpl::relay(protocol::TMExportShares& m) +{ + auto const shares = detail::parseExportShareBatch(m); + if (!shares) + { + JLOG(journal_.error()) + << "Refusing to relay malformed ExportShare batch"; + return; + } + + using Route = std::pair>; + std::vector routes; + routes.reserve(shares->size()); + for (std::size_t i = 0; i < shares->size(); ++i) + { + if (auto const toSkip = + app_.getHashRouter().shouldRelay((*shares)[i].wireHash())) + routes.emplace_back(i, std::move(*toSkip)); + } + + if (routes.empty()) + return; + + for_each([&](std::shared_ptr&& peer) { + protocol::TMExportShares outbound; + outbound.mutable_shares()->Reserve(routes.size()); + for (auto const& [index, toSkip] : routes) + { + if (!toSkip.contains(peer->id())) + outbound.add_shares(m.shares(index)); + } + + if (outbound.shares_size() != 0) + peer->send( + std::make_shared(outbound, protocol::mtEXPORT_SHARES)); + }); +} + +void +OverlayImpl::setExportShareHandler(ExportShareHandler handler) +{ + std::lock_guard lock(exportShareHandlerLock_); + exportShareHandler_ = std::move(handler); +} + +bool +OverlayImpl::acceptExportShare(ExportShare const& share) +{ + ExportShareHandler handler; + { + std::lock_guard lock(exportShareHandlerLock_); + handler = exportShareHandler_; + } + + if (!handler) + return false; + + try + { + return handler(share); + } + catch (std::exception const& e) + { + JLOG(journal_.error()) + << "ExportShare admission callback failed: " << e.what(); + } + catch (...) + { + JLOG(journal_.error()) << "ExportShare admission callback failed"; + } + return false; +} + std::set OverlayImpl::relay( protocol::TMValidation& m, diff --git a/src/xrpld/overlay/detail/OverlayImpl.h b/src/xrpld/overlay/detail/OverlayImpl.h index c5cfe6274..754397bed 100644 --- a/src/xrpld/overlay/detail/OverlayImpl.h +++ b/src/xrpld/overlay/detail/OverlayImpl.h @@ -131,6 +131,9 @@ private: // Protects the message and the sequence list of manifests std::mutex manifestLock_; + ExportShareHandler exportShareHandler_; + std::mutex exportShareHandlerLock_; + //-------------------------------------------------------------------------- public: @@ -223,6 +226,18 @@ public: void broadcast(protocol::TMValidation& m) override; + void + broadcast(protocol::TMExportShares& m) override; + + void + relay(protocol::TMExportShares& m) override; + + void + setExportShareHandler(ExportShareHandler handler) override; + + bool + acceptExportShare(ExportShare const& share) override; + std::set relay( protocol::TMProposeSet& m, diff --git a/src/xrpld/overlay/detail/PeerImp.cpp b/src/xrpld/overlay/detail/PeerImp.cpp index d4f9261b0..39db75af5 100644 --- a/src/xrpld/overlay/detail/PeerImp.cpp +++ b/src/xrpld/overlay/detail/PeerImp.cpp @@ -1126,6 +1126,57 @@ PeerImp::onMessage(std::shared_ptr const& m) }); } +void +PeerImp::onMessage(std::shared_ptr const& m) +{ + auto shares = detail::parseExportShareBatch(*m); + if (!shares) + { + fee_.update(Resource::feeMalformedRequest, "malformed export shares"); + return; + } + + std::vector fresh; + fresh.reserve(shares->size()); + for (std::size_t i = 0; i < shares->size(); ++i) + { + // This hash identifies only identical canonical wire frames. Semantic + // contribution deduplication belongs to application admission. + if (app_.getHashRouter().addSuppressionPeer( + (*shares)[i].wireHash(), id_)) + fresh.push_back(i); + } + + if (fresh.empty()) + { + fee_.update(Resource::feeUselessData, "duplicate export shares"); + return; + } + + std::weak_ptr weak = shared_from_this(); + app_.getJobQueue().addJob( + jtPEER, + "recvExportShares", + [weak, m, shares = std::move(*shares), fresh = std::move(fresh)]() { + auto const peer = weak.lock(); + if (!peer) + return; + + protocol::TMExportShares accepted; + accepted.mutable_shares()->Reserve(fresh.size()); + for (auto const index : fresh) + { + if (peer->overlay_.acceptExportShare(shares[index])) + accepted.add_shares(m->shares(index)); + } + + // Structural validity is insufficient: only application-admitted + // frames are eligible for relay. + if (accepted.shares_size() != 0) + peer->overlay_.relay(accepted); + }); +} + void PeerImp::onMessage(std::shared_ptr const& m) { diff --git a/src/xrpld/overlay/detail/PeerImp.h b/src/xrpld/overlay/detail/PeerImp.h index d77e6ab6e..5a12cec20 100644 --- a/src/xrpld/overlay/detail/PeerImp.h +++ b/src/xrpld/overlay/detail/PeerImp.h @@ -594,6 +594,8 @@ public: onMessage(std::shared_ptr const& m); void onMessage(std::shared_ptr const& m); + void + onMessage(std::shared_ptr const& m); private: //-------------------------------------------------------------------------- diff --git a/src/xrpld/overlay/detail/ProtocolMessage.h b/src/xrpld/overlay/detail/ProtocolMessage.h index 54f99eb73..e4360cafa 100644 --- a/src/xrpld/overlay/detail/ProtocolMessage.h +++ b/src/xrpld/overlay/detail/ProtocolMessage.h @@ -25,6 +25,8 @@ #include #include #include +#include +#include #include #include #include @@ -94,6 +96,8 @@ protocolMessageName(int type) return "have_transactions"; case protocol::mtTRANSACTIONS: return "transactions"; + case protocol::mtEXPORT_SHARES: + return "export_shares"; case protocol::mtSQUELCH: return "squelch"; case protocol::mtPROOF_PATH_REQ: @@ -112,6 +116,45 @@ protocolMessageName(int type) namespace detail { +/** Parse and structurally validate a canonical ExportShare relay batch. + + maxExportShareRelayPayloadBytes applies to the sum of repeated-field value + bytes. maxExportShareRelayMessageBytes additionally includes protobuf tags + and length prefixes. Neither limit includes the overlay message header. +*/ +inline std::optional> +parseExportShareBatch(protocol::TMExportShares const& message) +{ + auto const count = message.shares_size(); + if (count <= 0 || + static_cast(count) > + ExportLimits::maxExportSharesPerRelay || + message.ByteSizeLong() > ExportLimits::maxExportShareRelayMessageBytes) + return std::nullopt; + + std::size_t payloadBytes = 0; + std::vector result; + result.reserve(count); + + for (auto const& bytes : message.shares()) + { + if (bytes.empty() || + bytes.size() > ExportLimits::maxSerializedExportShareBytes || + payloadBytes > + ExportLimits::maxExportShareRelayPayloadBytes - bytes.size()) + return std::nullopt; + + payloadBytes += bytes.size(); + auto share = ExportShare::parse(makeSlice(bytes)); + if (!share || share->serialize().slice() != makeSlice(bytes)) + return std::nullopt; + + result.push_back(std::move(*share)); + } + + return result; +} + struct MessageHeader { /** The size of the message on the wire. @@ -450,6 +493,10 @@ invokeProtocolMessage( success = detail::invoke( *header, buffers, handler); break; + case protocol::mtEXPORT_SHARES: + success = detail::invoke( + *header, buffers, handler); + break; case protocol::mtSQUELCH: success = detail::invoke(*header, buffers, handler); diff --git a/src/xrpld/overlay/detail/TrafficCount.cpp b/src/xrpld/overlay/detail/TrafficCount.cpp index c64a033e3..43f07e079 100644 --- a/src/xrpld/overlay/detail/TrafficCount.cpp +++ b/src/xrpld/overlay/detail/TrafficCount.cpp @@ -157,6 +157,9 @@ TrafficCount::categorize( if (type == protocol::mtTRANSACTIONS) return TrafficCount::category::requested_transactions; + if (type == protocol::mtEXPORT_SHARES) + return TrafficCount::category::export_shares; + return TrafficCount::category::unknown; } diff --git a/src/xrpld/overlay/detail/TrafficCount.h b/src/xrpld/overlay/detail/TrafficCount.h index 9b4fabdc3..44be9ea30 100644 --- a/src/xrpld/overlay/detail/TrafficCount.h +++ b/src/xrpld/overlay/detail/TrafficCount.h @@ -157,6 +157,9 @@ public: // TMTransactions requested_transactions, + // TMExportShares + export_shares, + unknown // must be last }; @@ -249,6 +252,7 @@ protected: {"replay_delta_response"}, // category::replay_delta_response {"have_transactions"}, // category::have_transactions {"requested_transactions"}, // category::transactions + {"export_shares"}, // category::export_shares {"unknown"} // category::unknown }}; };