xrpld
Loading...
Searching...
No Matches
OverlayImpl.cpp
1#include <xrpld/overlay/detail/OverlayImpl.h>
2
3#include <xrpld/app/misc/ValidatorList.h>
4#include <xrpld/app/misc/ValidatorSite.h>
5#include <xrpld/overlay/Cluster.h>
6#include <xrpld/overlay/detail/ConnectAttempt.h>
7#include <xrpld/overlay/detail/Handshake.h>
8#include <xrpld/overlay/detail/PeerImp.h>
9#include <xrpld/overlay/detail/ProtocolVersion.h>
10#include <xrpld/overlay/detail/TrafficCount.h>
11#include <xrpld/overlay/detail/Tuning.h>
12#include <xrpld/peerfinder/PeerfinderManager.h>
13#include <xrpld/rpc/ServerHandler.h>
14#include <xrpld/rpc/handlers/admin/status/GetCounts.h>
15#include <xrpld/rpc/json_body.h>
16
17#include <xrpl/basics/Log.h>
18#include <xrpl/basics/Resolver.h>
19#include <xrpl/basics/Slice.h>
20#include <xrpl/basics/base64.h>
21#include <xrpl/basics/base_uint.h>
22#include <xrpl/basics/chrono.h>
23#include <xrpl/basics/contract.h>
24#include <xrpl/basics/make_SSLContext.h>
25#include <xrpl/basics/random.h>
26#include <xrpl/basics/strHex.h>
27#include <xrpl/beast/core/LexicalCast.h>
28#include <xrpl/beast/insight/Collector.h>
29#include <xrpl/beast/net/IPAddress.h>
30#include <xrpl/beast/net/IPAddressConversion.h>
31#include <xrpl/beast/net/IPEndpoint.h>
32#include <xrpl/beast/rfc2616.h>
33#include <xrpl/beast/utility/PropertyStream.h>
34#include <xrpl/beast/utility/WrappedSink.h>
35#include <xrpl/beast/utility/instrumentation.h>
36#include <xrpl/config/BasicConfig.h>
37#include <xrpl/config/Constants.h>
38#include <xrpl/core/HashRouter.h>
39#include <xrpl/json/json_value.h>
40#include <xrpl/peerfinder/Config.h>
41#include <xrpl/peerfinder/Slot.h>
42#include <xrpl/peerfinder/make_Manager.h>
43#include <xrpl/protocol/BuildInfo.h>
44#include <xrpl/protocol/STTx.h>
45#include <xrpl/protocol/Serializer.h>
46#include <xrpl/protocol/SystemParameters.h>
47#include <xrpl/protocol/jss.h>
48#include <xrpl/resource/Fees.h>
49#include <xrpl/resource/ResourceManager.h>
50#include <xrpl/server/Handoff.h>
51#include <xrpl/server/Manifest.h>
52#include <xrpl/server/NetworkOPs.h>
53#include <xrpl/server/SimpleWriter.h>
54#include <xrpl/server/Wallet.h>
55#include <xrpl/server/Writer.h>
56
57#include <boost/algorithm/string/predicate.hpp>
58#include <boost/asio/bind_executor.hpp>
59#include <boost/asio/dispatch.hpp>
60#include <boost/asio/error.hpp>
61#include <boost/asio/executor_work_guard.hpp>
62#include <boost/asio/io_context.hpp>
63#include <boost/asio/ip/address.hpp>
64#include <boost/asio/post.hpp>
65#include <boost/asio/strand.hpp>
66#include <boost/beast/http/empty_body.hpp>
67#include <boost/beast/http/field.hpp>
68#include <boost/beast/http/status.hpp>
69#include <boost/lexical_cast.hpp>
70#include <boost/lexical_cast/bad_lexical_cast.hpp>
71#include <boost/lexical_cast/try_lexical_convert.hpp>
72
73#include <xrpl.pb.h>
74
75#include <algorithm>
76#include <chrono>
77#include <cstddef>
78#include <cstdint>
79#include <exception>
80#include <functional>
81#include <iomanip>
82#include <memory>
83#include <mutex>
84#include <optional>
85#include <set>
86#include <sstream>
87#include <stdexcept>
88#include <string>
89#include <string_view>
90#include <tuple>
91#include <unordered_map>
92#include <utility>
93#include <vector>
94
95namespace xrpl {
96
97namespace crawl_options {
98static constexpr auto kDisabled = 0;
99static constexpr auto kOverlay = (1 << 0);
100static constexpr auto kServerInfo = (1 << 1);
101static constexpr auto kServerCounts = (1 << 2);
102static constexpr auto kUnl = (1 << 3);
103} // namespace crawl_options
104
105//------------------------------------------------------------------------------
106
108{
109}
110
112{
113 overlay_.remove(*this);
114}
115
116//------------------------------------------------------------------------------
117
121
122void
124{
125 // This method is only ever called from the same strand that calls
126 // Timer::on_timer, ensuring they never execute concurrently.
127 stopping = true;
128 timer.cancel();
129}
130
131void
133{
134 timer.expires_after(std::chrono::seconds(1));
135 timer.async_wait(
136 boost::asio::bind_executor(
137 overlay_.strand_,
138 [self = shared_from_this()](error_code const& ec) { self->onTimer(ec); }));
139}
140
141void
143{
144 if (ec || stopping)
145 {
146 if (ec && ec != boost::asio::error::operation_aborted)
147 {
148 JLOG(overlay_.journal_.error()) << "on_timer: " << ec.message();
149 }
150 return;
151 }
152
153 overlay_.peerFinder_->oncePerSecond();
154 overlay_.sendEndpoints();
155 overlay_.autoConnect();
156 if (overlay_.app_.config().txReduceRelayEnable)
157 overlay_.sendTxQueue();
158
159 if ((++overlay_.timerCount_ % tuning::kCheckIdlePeers) == 0)
160 overlay_.deleteIdlePeers();
161
162 asyncWait();
163}
164
165//------------------------------------------------------------------------------
166
168 Application& app,
169 Setup setup,
170 ServerHandler& serverHandler,
172 Resolver& resolver,
173 boost::asio::io_context& ioContext,
174 BasicConfig const& config,
175 beast::insight::Collector::ptr const& collector)
176 : app_(app)
177 , ioContext_(ioContext)
178 , work_(std::in_place, boost::asio::make_work_guard(ioContext_))
179 , strand_(boost::asio::make_strand(ioContext_))
180 , setup_(std::move(setup))
181 , journal_(app_.getJournal("Overlay"))
182 , serverHandler_(serverHandler)
184 , store_(app_.getJournal("PeerFinder"))
185 , peerFinder_(
186 peer_finder::makeManager(
187 ioContext,
188 stopwatch(),
189 app_.getJournal("PeerFinder"),
190 store_,
191 collector))
192 , resolver_(resolver)
193 , nextId_(1)
194 , slots_(app, *this, app.config())
195 , stats_(
196 [this] { collectMetrics(); },
197 collector,
198 [counts = traffic_.getCounts(), collector]() {
200
201 for (auto const& pair : counts)
202 ret.emplace(pair.first, TrafficGauges(pair.second.name, collector));
203
204 return ret;
205 }())
206{
207 store_.open(config);
208 beast::PropertyStream::Source::add(peerFinder_.get());
209}
210
214 http_request_type&& request,
215 endpoint_type remoteEndpoint)
216{
217 auto const id = nextId_++;
218 auto peerJournal = app_.getJournal("Peer");
219 beast::WrappedSink sink(peerJournal.sink(), makePrefix(id));
220 beast::Journal const journal(sink);
221
222 Handoff handoff;
223 if (processRequest(request, handoff))
224 return handoff;
225 if (!isPeerUpgrade(request))
226 return handoff;
227
228 handoff.moved = true;
229
230 JLOG(journal.debug()) << "Peer connection upgrade from " << remoteEndpoint;
231
232 error_code ec;
233 auto const localEndpoint(streamPtr->next_layer().socket().local_endpoint(ec));
234 if (ec)
235 {
236 JLOG(journal.debug()) << remoteEndpoint << " failed: " << ec.message();
237 return handoff;
238 }
239
240 auto consumer =
241 resourceManager_.newInboundEndpoint(beast::IPAddressConversion::fromAsio(remoteEndpoint));
242 if (consumer.disconnect(journal))
243 return handoff;
244
245 auto const [slot, result] = peerFinder_->newInboundSlot(
248
249 if (slot == nullptr)
250 {
251 // connection refused either IP limit exceeded or self-connect
252 handoff.moved = false;
253 JLOG(journal.debug()) << "Peer " << remoteEndpoint << " refused, " << to_string(result);
254 return handoff;
255 }
256
257 // Validate HTTP request
258
259 {
260 auto const types = beast::rfc2616::splitCommas(request["Connect-As"]);
261 if (std::ranges::find_if(types, [](std::string const& s) {
262 return boost::iequals(s, "peer");
263 }) == types.end())
264 {
265 handoff.moved = false;
266 handoff.response = makeRedirectResponse(slot, request, remoteEndpoint.address());
267 handoff.keepAlive = beast::rfc2616::isKeepAlive(request);
268 return handoff;
269 }
270 }
271
272 auto const negotiatedVersion = negotiateProtocolVersion(request["Upgrade"]);
273 if (!negotiatedVersion)
274 {
275 peerFinder_->onClosed(slot);
276 handoff.moved = false;
277 handoff.response = makeErrorResponse(
278 slot, request, remoteEndpoint.address(), "Unable to agree on a protocol version");
279 handoff.keepAlive = false;
280 return handoff;
281 }
282
283 auto const sharedValue = makeSharedValue(*streamPtr, journal);
284 if (!sharedValue)
285 {
286 peerFinder_->onClosed(slot);
287 handoff.moved = false;
288 handoff.response =
289 makeErrorResponse(slot, request, remoteEndpoint.address(), "Incorrect security cookie");
290 handoff.keepAlive = false;
291 return handoff;
292 }
293
294 try
295 {
296 auto publicKey = verifyHandshake(
297 request,
298 *sharedValue,
299 setup_.networkID,
300 setup_.publicIp,
301 remoteEndpoint.address(),
302 app_);
303
304 consumer.setPublicKey(publicKey);
305
306 {
307 // The node gets a reserved slot if it is in our cluster
308 // or if it has a reservation.
309 bool const reserved = static_cast<bool>(app_.getCluster().member(publicKey)) ||
310 app_.getPeerReservations().contains(publicKey);
311 auto const result = peerFinder_->activate(slot, publicKey, reserved);
312 if (result != peer_finder::Result::Success)
313 {
314 peerFinder_->onClosed(slot);
315 JLOG(journal.debug())
316 << "Peer " << remoteEndpoint << " redirected, " << to_string(result);
317 handoff.moved = false;
318 handoff.response = makeRedirectResponse(slot, request, remoteEndpoint.address());
319 handoff.keepAlive = false;
320 return handoff;
321 }
322 }
323
324 auto const peer = std::make_shared<PeerImp>(
325 app_,
326 id,
327 slot,
328 std::move(request),
329 publicKey,
330 *negotiatedVersion,
331 consumer,
332 std::move(streamPtr),
333 *this);
334 {
335 // As we are not on the strand, run() must be called
336 // while holding the lock, otherwise new I/O can be
337 // queued after a call to stop().
338 std::scoped_lock const lock(mutex_);
339 {
340 auto const result = peers_.emplace(peer->slot(), peer);
341 XRPL_ASSERT(result.second, "xrpl::OverlayImpl::onHandoff : peer is inserted");
342 (void)result.second;
343 }
344 list_.emplace(peer.get(), peer);
345
346 peer->run();
347 }
348 handoff.moved = true;
349 return handoff;
350 }
351 catch (std::exception const& e)
352 {
353 JLOG(journal.debug()) << "Peer " << remoteEndpoint << " fails handshake (" << e.what()
354 << ")";
355
356 peerFinder_->onClosed(slot);
357 handoff.moved = false;
358 handoff.response = makeErrorResponse(slot, request, remoteEndpoint.address(), e.what());
359 handoff.keepAlive = false;
360 return handoff;
361 }
362}
363
364//------------------------------------------------------------------------------
365
366bool
368{
369 if (!isUpgrade(request))
370 return false;
371 auto const versions = parseProtocolVersions(request["Upgrade"]);
372 return !versions.empty();
373}
374
377{
379 ss << "[" << std::setfill('0') << std::setw(3) << id << "] ";
380 return ss.str();
381}
382
386 http_request_type const& request,
387 address_type remoteAddress)
388{
389 boost::beast::http::response<JsonBody> msg;
390 msg.version(request.version());
391 msg.result(boost::beast::http::status::service_unavailable);
392 msg.insert("Server", build_info::getFullVersionString());
393 {
395 ostr << remoteAddress;
396 msg.insert("Remote-Address", ostr.str());
397 }
398 msg.insert("Content-Type", "application/json");
399 msg.insert(boost::beast::http::field::connection, "close");
400 msg.body() = json::ValueType::Object;
401 {
402 json::Value& ips = (msg.body()["peer-ips"] = json::ValueType::Array);
403 for (auto const& _ : peerFinder_->redirect(slot))
404 ips.append(_.address.toString());
405 }
406 msg.prepare_payload();
408}
409
413 http_request_type const& request,
414 address_type remoteAddress,
415 std::string const& text)
416{
417 boost::beast::http::response<boost::beast::http::empty_body> msg;
418 msg.version(request.version());
419 msg.result(boost::beast::http::status::bad_request);
420 msg.reason("Bad Request (" + text + ")");
421 msg.insert("Server", build_info::getFullVersionString());
422 msg.insert("Remote-Address", remoteAddress.to_string());
423 msg.insert(boost::beast::http::field::connection, "close");
424 msg.prepare_payload();
426}
427
428//------------------------------------------------------------------------------
429
430void
432{
433 XRPL_ASSERT(work_, "xrpl::OverlayImpl::connect : work is set");
434
435 auto usage = resourceManager().newOutboundEndpoint(remoteEndpoint);
436 if (usage.disconnect(journal_))
437 {
438 JLOG(journal_.info()) << "Over resource limit: " << remoteEndpoint;
439 return;
440 }
441
442 auto const [slot, result] = peerFinder().newOutboundSlot(remoteEndpoint);
443 if (slot == nullptr)
444 {
445 JLOG(journal_.debug()) << "Connect: No slot for " << remoteEndpoint << ": "
446 << to_string(result);
447 return;
448 }
449
451 app_,
454 usage,
455 setup_.context,
456 nextId_++,
457 slot,
458 app_.getJournal("Peer"),
459 *this);
460
461 std::scoped_lock const lock(mutex_);
462 list_.emplace(p.get(), p);
463 p->run();
464}
465
466//------------------------------------------------------------------------------
467
468// Adds a peer that is already handshaked and active
469void
471{
472 beast::WrappedSink sink{journal_.sink(), peer->prefix()};
473 beast::Journal const journal{sink};
474
475 std::scoped_lock const lock(mutex_);
476
477 {
478 auto const result = peers_.emplace(peer->slot(), peer);
479 XRPL_ASSERT(result.second, "xrpl::OverlayImpl::addActive : peer is inserted");
480 (void)result.second;
481 }
482
483 {
484 auto const result = ids_.emplace(
486 XRPL_ASSERT(result.second, "xrpl::OverlayImpl::addActive : peer ID is inserted");
487 (void)result.second;
488 }
489
490 list_.emplace(peer.get(), peer);
491
492 JLOG(journal.debug()) << "activated";
493
494 // As we are not on the strand, run() must be called
495 // while holding the lock, otherwise new I/O can be
496 // queued after a call to stop().
497 peer->run();
498}
499
500void
502{
503 std::scoped_lock const lock(mutex_);
504 auto const iter = peers_.find(slot);
505 XRPL_ASSERT(iter != peers_.end(), "xrpl::OverlayImpl::remove : valid input");
506 peers_.erase(iter);
507}
508
509void
511{
513 app_.config(),
514 serverHandler_.setup().overlay.port(),
515 app_.getValidationPublicKey().has_value(),
516 setup_.ipLimit,
517 setup_.verifyEndpoints);
518
519 peerFinder_->setConfig(config);
520 peerFinder_->start();
521
522 // Populate our boot cache: if there are no entries in [ips] then we use
523 // the entries in [ips_fixed].
524 auto bootstrapIps = app_.config().ips.empty() ? app_.config().ipsFixed : app_.config().ips;
525
526 // If nothing is specified, default to several well-known high-capacity
527 // servers to serve as bootstrap:
528 if (bootstrapIps.empty())
529 {
530 // Pool of servers operated by Ripple Labs Inc. - https://ripple.com
531 bootstrapIps.emplace_back("r.ripple.com 51235");
532
533 // Pool of servers operated by ISRDC - https://isrdc.in
534 bootstrapIps.emplace_back("sahyadri.isrdc.in 51235");
535
536 // Pool of servers operated by @Xrpkuwait - https://xrpkuwait.com
537 bootstrapIps.emplace_back("hubs.xrpkuwait.com 51235");
538
539 // Pool of servers operated by XRPL Commons - https://xrpl-commons.org
540 bootstrapIps.emplace_back("hub.xrpl-commons.org 51235");
541 }
542
543 resolver_.resolve(
544 bootstrapIps,
545 [this](std::string const& name, std::vector<beast::ip::Endpoint> const& addresses) {
547 ips.reserve(addresses.size());
548 for (auto const& addr : addresses)
549 {
550 if (addr.port() == 0)
551 {
552 ips.push_back(to_string(addr.atPort(kDefaultPeerPort)));
553 }
554 else
555 {
556 ips.push_back(to_string(addr));
557 }
558 }
559
560 std::string const base("config: ");
561 if (!ips.empty())
562 peerFinder_->addFallbackStrings(base + name, ips);
563 });
564
565 // Add the ips_fixed from the xrpld.cfg file
566 if (!app_.config().standalone() && !app_.config().ipsFixed.empty())
567 {
568 resolver_.resolve(
569 app_.config().ipsFixed,
570 [this](std::string const& name, std::vector<beast::ip::Endpoint> const& addresses) {
571 std::vector<beast::ip::Endpoint> ips;
572 ips.reserve(addresses.size());
573
574 for (auto& addr : addresses)
575 {
576 if (addr.port() == 0)
577 {
578 ips.emplace_back(addr.address(), kDefaultPeerPort);
579 }
580 else
581 {
582 ips.emplace_back(addr);
583 }
584 }
585
586 if (!ips.empty())
587 peerFinder_->addFixedPeer(name, ips);
588 });
589 }
590 auto const timer = std::make_shared<Timer>(*this);
591 std::scoped_lock const lock(mutex_);
592 list_.emplace(timer.get(), timer);
593 timer_ = timer;
594 timer->asyncWait();
595}
596
597void
599{
600 boost::asio::dispatch(strand_, [this] { stopChildren(); });
601 {
602 std::unique_lock<decltype(mutex_)> lock(mutex_);
603 cond_.wait(lock, [this] { return list_.empty(); });
604 }
605 peerFinder_->stop();
606}
607
608//------------------------------------------------------------------------------
609//
610// PropertyStream
611//
612//------------------------------------------------------------------------------
613
614void
616{
617 beast::PropertyStream::Set set("traffic", stream);
618 auto const stats = traffic_.getCounts();
619 for (auto const& pair : stats)
620 {
622 item["category"] = pair.second.name;
623 item["bytes_in"] = std::to_string(pair.second.bytesIn.load());
624 item["messages_in"] = std::to_string(pair.second.messagesIn.load());
625 item["bytes_out"] = std::to_string(pair.second.bytesOut.load());
626 item["messages_out"] = std::to_string(pair.second.messagesOut.load());
627 }
628}
629
630//------------------------------------------------------------------------------
637void
639{
640 beast::WrappedSink sink{journal_.sink(), peer->prefix()};
641 beast::Journal const journal{sink};
642
643 // Now track this peer
644 {
645 std::scoped_lock const lock(mutex_);
646 auto const result(ids_.emplace(
648 XRPL_ASSERT(result.second, "xrpl::OverlayImpl::activate : peer ID is inserted");
649 (void)result.second;
650 }
651
652 JLOG(journal.debug()) << "activated";
653
654 // We just accepted this peer so we have non-zero active peers
655 XRPL_ASSERT(size(), "xrpl::OverlayImpl::activate : nonzero peers");
656}
657
658void
660{
661 std::scoped_lock const lock(mutex_);
662 ids_.erase(id);
663}
664
665void
668 std::shared_ptr<PeerImp> const& from)
669{
670 auto const& journal = from->pJournal();
671
672 // Process every trusted manifest, but stop processing untrusted ones once
673 // the configured untrusted count has been handled, so the work stays
674 // bounded. Trusted manifests are always processed: dropping one would delay
675 // a validator key rotation reaching this node.
676 auto const maxUntrusted = untrustedManifestCount(app_.config().maxUntrustedCount);
677 auto const total = static_cast<std::size_t>(m->list_size());
678 std::size_t untrusted = 0;
679 bool skippedUntrusted = false;
680
681 protocol::TMManifests relay;
682
683 for (std::size_t i = 0; i < total; ++i)
684 {
685 auto& s = m->list().Get(i).stobject();
686
687 if (auto mo = deserializeManifest(s))
688 {
689 auto const serialized = mo->serialized;
690 // Resolve trust before applyManifest takes the manifest-cache
691 // lock: listed() takes the validator-list lock, so ordering it
692 // first avoids holding the two locks in opposite orders.
693 bool const isTrusted = app_.getValidators().listed(mo->masterKey);
694
695 // Bound untrusted work: process at most maxUntrusted untrusted
696 // manifests, but never skip a trusted one. Trusted manifests are
697 // not counted against the cap.
698 if (!isTrusted)
699 {
700 if (untrusted >= maxUntrusted)
701 {
702 skippedUntrusted = true;
703 continue;
704 }
705 ++untrusted;
706 }
707
708 auto const result = app_.getValidatorManifests().applyManifest(
709 std::move(*mo),
712
713 if (result == ManifestDisposition::Accepted)
714 {
715 // N.B.: this is important; the applyManifest call above moves
716 // the loaded Manifest out of the optional so we need to
717 // reload it here.
718 mo = deserializeManifest(serialized);
719 XRPL_ASSERT(
720 mo,
721 "xrpl::OverlayImpl::onManifests : manifest "
722 "deserialization succeeded");
723 // NOLINTBEGIN(bugprone-unchecked-optional-access) assert above
724 app_.getOPs().pubManifest(*mo);
725 // NOLINTEND(bugprone-unchecked-optional-access)
726
727 relay.add_list()->set_stobject(s);
728
729 // Persist to the wallet DB only for trusted keys, so untrusted
730 // gossip never survives a restart.
731 if (isTrusted)
732 {
733 auto db = app_.getWalletDB().checkoutDb();
734 addValidatorManifest(*db, serialized);
735 }
736 }
737 }
738 else
739 {
740 JLOG(journal.debug()) << "Malformed manifest #" << i + 1 << ": " << strHex(s);
741 continue;
742 }
743 }
744
745 if (skippedUntrusted)
746 {
747 // The sender exceeded the untrusted per-message cap. Charge it (once,
748 // here) so a flood of untrusted manifests is penalized.
749 from->charge(resource::kFeeMalformedRequest, "too many untrusted manifests");
750
751 JLOG(journal.warn()) << "Manifests: message had " << total
752 << " entries; processed all trusted plus the first " << maxUntrusted
753 << " untrusted";
754 }
755
756 if (!relay.list().empty())
757 {
758 forEach([m2 = std::make_shared<Message>(relay, protocol::mtMANIFESTS)](
759 std::shared_ptr<PeerImp> const& p) { p->send(m2); });
760 }
761}
762
763void
765{
766 traffic_.addCount(cat, true, size);
767}
768
769void
771{
772 traffic_.addCount(cat, false, size);
773}
774
781{
782 std::scoped_lock const lock(mutex_);
783 return ids_.size();
784}
785
786int
788{
789 return peerFinder_->config().maxPeers;
790}
791
794{
795 using namespace std::chrono;
796 json::Value jv;
797 auto& av = jv[jss::active] = json::Value(json::ValueType::Array);
798
799 forEach([&](std::shared_ptr<PeerImp> const& sp) {
800 auto& pv = av.append(json::Value(json::ValueType::Object));
801 pv[jss::public_key] = base64Encode(sp->getNodePublic().data(), sp->getNodePublic().size());
802 pv[jss::type] = sp->slot()->inbound() ? jss::in : jss::out;
803 pv[jss::uptime] = static_cast<std::uint32_t>(duration_cast<seconds>(sp->uptime()).count());
804 if (sp->crawl())
805 {
806 pv[jss::ip] = sp->getRemoteAddress().address().to_string();
807 if (sp->slot()->inbound())
808 {
809 if (auto port = sp->slot()->listeningPort())
810 pv[jss::port] = *port;
811 }
812 else
813 {
814 pv[jss::port] = sp->getRemoteAddress().port();
815 }
816 }
817
818 {
819 auto version{sp->getVersion()};
820 if (!version.empty())
821 {
822 // Could move here if json::value supported moving from strings
823 pv[jss::version] = std::string{version};
824 }
825 }
826
827 std::uint32_t minSeq = 0, maxSeq = 0;
828 sp->ledgerRange(minSeq, maxSeq);
829 if (minSeq != 0 || maxSeq != 0)
830 pv[jss::complete_ledgers] = std::to_string(minSeq) + "-" + std::to_string(maxSeq);
831 });
832
833 return jv;
834}
835
838{
839 bool const humanReadable = false;
840 bool const admin = false;
841 bool const counters = false;
842
843 json::Value serverInfo = app_.getOPs().getServerInfo(humanReadable, admin, counters);
844
845 // Filter out some information
846 serverInfo.removeMember(jss::hostid);
847 serverInfo.removeMember(jss::load_factor_fee_escalation);
848 serverInfo.removeMember(jss::load_factor_fee_queue);
849 serverInfo.removeMember(jss::validation_quorum);
850
851 if (serverInfo.isMember(jss::validated_ledger))
852 {
853 json::Value& validatedLedger = serverInfo[jss::validated_ledger];
854
855 validatedLedger.removeMember(jss::base_fee);
856 validatedLedger.removeMember(jss::reserve_base_xrp);
857 validatedLedger.removeMember(jss::reserve_inc_xrp);
858 }
859
860 return serverInfo;
861}
862
868
871{
872 json::Value validators = app_.getValidators().getJson();
873
874 if (validators.isMember(jss::publisher_lists))
875 {
876 json::Value& publisherLists = validators[jss::publisher_lists];
877
878 for (auto& publisher : publisherLists)
879 {
880 publisher.removeMember(jss::list);
881 }
882 }
883
884 validators.removeMember(jss::signing_keys);
885 validators.removeMember(jss::trusted_validator_keys);
886 validators.removeMember(jss::validation_quorum);
887
888 json::Value validatorSites = app_.getValidatorSites().getJson();
889
890 if (validatorSites.isMember(jss::validator_sites))
891 {
892 validators[jss::validator_sites] = std::move(validatorSites[jss::validator_sites]);
893 }
894
895 return validators;
896}
897
898// Returns information on verified peers.
901{
903 for (auto const& peer : getActivePeers())
904 {
905 json.append(peer->json());
906 }
907 return json;
908}
909
910bool
912{
913 if (req.target() != "/crawl" || setup_.crawlOptions == crawl_options::kDisabled)
914 return false;
915
916 boost::beast::http::response<JsonBody> msg;
917 msg.version(req.version());
918 msg.result(boost::beast::http::status::ok);
919 msg.insert("Server", build_info::getFullVersionString());
920 msg.insert("Content-Type", "application/json");
921 msg.insert("Connection", "close");
922 msg.body()["version"] = json::Value(2u);
923
924 if ((setup_.crawlOptions & crawl_options::kOverlay) != 0u)
925 {
926 msg.body()["overlay"] = getOverlayInfo();
927 }
928 if ((setup_.crawlOptions & crawl_options::kServerInfo) != 0u)
929 {
930 msg.body()["server"] = getServerInfo();
931 }
932 if ((setup_.crawlOptions & crawl_options::kServerCounts) != 0u)
933 {
934 msg.body()["counts"] = getServerCounts();
935 }
936 if ((setup_.crawlOptions & crawl_options::kUnl) != 0u)
937 {
938 msg.body()["unl"] = getUnlInfo();
939 }
940
941 msg.prepare_payload();
943 return true;
944}
945
946bool
948{
949 // If the target is in the form "/vl/<validator_list_public_key>",
950 // return the most recent validator list for that key.
951 constexpr std::string_view kPrefix("/vl/");
952
953 if (!req.target().starts_with(kPrefix) || !setup_.vlEnabled)
954 return false;
955
956 std::uint32_t version = 1;
957
958 boost::beast::http::response<JsonBody> msg;
959 msg.version(req.version());
960 msg.insert("Server", build_info::getFullVersionString());
961 msg.insert("Content-Type", "application/json");
962 msg.insert("Connection", "close");
963
964 auto fail = [&msg, &handoff](auto status) {
965 msg.result(status);
966 msg.insert("Content-Length", "0");
967
968 msg.body() = json::ValueType::Null;
969
970 msg.prepare_payload();
972 return true;
973 };
974
975 std::string_view key = req.target().substr(kPrefix.size());
976
977 if (auto slash = key.find('/'); slash != std::string_view::npos)
978 {
979 auto verString = key.substr(0, slash);
980 if (!boost::conversion::try_lexical_convert(verString, version))
981 return fail(boost::beast::http::status::bad_request);
982 key = key.substr(slash + 1);
983 }
984
985 if (key.empty())
986 return fail(boost::beast::http::status::bad_request);
987
988 // find the list
989 auto vl = app_.getValidators().getAvailable(key, version);
990
991 if (!vl)
992 {
993 // 404 not found
994 return fail(boost::beast::http::status::not_found);
995 }
996 if (!*vl)
997 {
998 return fail(boost::beast::http::status::bad_request);
999 }
1000
1001 msg.result(boost::beast::http::status::ok);
1002
1003 msg.body() = *vl;
1004
1005 msg.prepare_payload();
1007 return true;
1008}
1009
1010bool
1012{
1013 if (req.target() != "/health")
1014 return false;
1015 boost::beast::http::response<JsonBody> msg;
1016 msg.version(req.version());
1017 msg.insert("Server", build_info::getFullVersionString());
1018 msg.insert("Content-Type", "application/json");
1019 msg.insert("Connection", "close");
1020
1021 auto info = getServerInfo();
1022
1023 int lastValidatedLedgerAge = -1;
1024 if (info.isMember(jss::validated_ledger))
1025 lastValidatedLedgerAge = info[jss::validated_ledger][jss::age].asInt();
1026 bool amendmentBlocked = false;
1027 if (info.isMember(jss::amendment_blocked))
1028 amendmentBlocked = true;
1029 int const numberPeers = info[jss::peers].asInt();
1030 std::string const serverState = info[jss::server_state].asString();
1031 auto loadFactor = info[jss::load_factor_server].asDouble() / info[jss::load_base].asDouble();
1032
1033 enum class HealthState { Healthy, Warning, Critical };
1034 auto health = HealthState::Healthy;
1035 auto setHealth = [&health](HealthState state) { health = std::max(health, state); };
1036
1037 msg.body()[jss::info] = json::ValueType::Object;
1038 if (lastValidatedLedgerAge >= 7 || lastValidatedLedgerAge < 0)
1039 {
1040 msg.body()[jss::info][jss::validated_ledger] = lastValidatedLedgerAge;
1041 if (lastValidatedLedgerAge < 20)
1042 {
1043 setHealth(HealthState::Warning);
1044 }
1045 else
1046 {
1047 setHealth(HealthState::Critical);
1048 }
1049 }
1050
1051 if (amendmentBlocked)
1052 {
1053 msg.body()[jss::info][jss::amendment_blocked] = true;
1054 setHealth(HealthState::Critical);
1055 }
1056
1057 if (numberPeers <= 7)
1058 {
1059 msg.body()[jss::info][jss::peers] = numberPeers;
1060 if (numberPeers != 0)
1061 {
1062 setHealth(HealthState::Warning);
1063 }
1064 else
1065 {
1066 setHealth(HealthState::Critical);
1067 }
1068 }
1069
1070 if (!(serverState == "full" || serverState == "validating" || serverState == "proposing"))
1071 {
1072 msg.body()[jss::info][jss::server_state] = serverState;
1073 if (serverState == "syncing" || serverState == "tracking" || serverState == "connected")
1074 {
1075 setHealth(HealthState::Warning);
1076 }
1077 else
1078 {
1079 setHealth(HealthState::Critical);
1080 }
1081 }
1082
1083 if (loadFactor > 100)
1084 {
1085 msg.body()[jss::info][jss::load_factor] = loadFactor;
1086 if (loadFactor < 1000)
1087 {
1088 setHealth(HealthState::Warning);
1089 }
1090 else
1091 {
1092 setHealth(HealthState::Critical);
1093 }
1094 }
1095
1096 switch (health)
1097 {
1098 case HealthState::Healthy:
1099 msg.result(boost::beast::http::status::ok);
1100 break;
1101 case HealthState::Warning:
1102 msg.result(boost::beast::http::status::service_unavailable);
1103 break;
1104 case HealthState::Critical:
1105 msg.result(boost::beast::http::status::internal_server_error);
1106 break;
1107 }
1108
1109 msg.prepare_payload();
1111 return true;
1112}
1113
1114bool
1116{
1117 // Take advantage of || short-circuiting
1118 return processCrawl(req, handoff) || processValidatorList(req, handoff) ||
1119 processHealth(req, handoff);
1120}
1121
1124{
1126 ret.reserve(size());
1127
1128 forEach([&ret](std::shared_ptr<PeerImp> const& sp) { ret.emplace_back(sp); });
1129
1130 return ret;
1131}
1132
1135 std::set<Peer::id_t> const& toSkip,
1136 std::size_t& active,
1137 std::size_t& disabled,
1138 std::size_t& enabledInSkip) const
1139{
1142
1143 active = ids_.size();
1144 disabled = enabledInSkip = 0;
1145 ret.reserve(ids_.size());
1146
1147 // NOTE The purpose of p is to delay the destruction of PeerImp
1149 for (auto& [id, w] : ids_)
1150 {
1151 if (p = w.lock(); p != nullptr)
1152 {
1153 bool const reduceRelayEnabled = p->txReduceRelayEnabled();
1154 // tx reduced relay feature disabled
1155 if (!reduceRelayEnabled)
1156 ++disabled;
1157
1158 if (!toSkip.contains(id))
1159 {
1160 ret.emplace_back(std::move(p));
1161 }
1162 else if (reduceRelayEnabled)
1163 {
1164 ++enabledInSkip;
1165 }
1166 }
1167 }
1168
1169 return ret;
1170}
1171
1172void
1174{
1175 forEach([index](std::shared_ptr<PeerImp> const& sp) { sp->checkTracking(index); });
1176}
1177
1180{
1182 auto const iter = ids_.find(id);
1183 if (iter != ids_.end())
1184 return iter->second.lock();
1185 return {};
1186}
1187
1188// A public key hash map was not used due to the peer connect/disconnect
1189// update overhead outweighing the performance of a small set linear search.
1192{
1194 // NOTE The purpose of peer is to delay the destruction of PeerImp
1196 for (auto const& e : ids_)
1197 {
1198 if (peer = e.second.lock(); peer != nullptr)
1199 {
1200 if (peer->getNodePublic() == pubKey)
1201 return peer;
1202 }
1203 }
1204 return {};
1205}
1206
1207void
1208OverlayImpl::broadcast(protocol::TMProposeSet const& m)
1209{
1210 auto const sm = std::make_shared<Message>(m, protocol::mtPROPOSE_LEDGER);
1211 forEach([&](std::shared_ptr<PeerImp> const& p) { p->send(sm); });
1212}
1213
1215OverlayImpl::relay(protocol::TMProposeSet const& m, uint256 const& uid, PublicKey const& validator)
1216{
1217 if (auto const toSkip = app_.getHashRouter().shouldRelay(uid))
1218 {
1219 auto const sm = std::make_shared<Message>(m, protocol::mtPROPOSE_LEDGER, validator);
1220 forEach([&](std::shared_ptr<PeerImp> const& p) {
1221 if (!toSkip->contains(p->id()))
1222 p->send(sm);
1223 });
1224 return *toSkip;
1225 }
1226 return {};
1227}
1228
1229void
1230OverlayImpl::broadcast(protocol::TMValidation const& m)
1231{
1232 auto const sm = std::make_shared<Message>(m, protocol::mtVALIDATION);
1233 forEach([sm](std::shared_ptr<PeerImp> const& p) { p->send(sm); });
1234}
1235
1237OverlayImpl::relay(protocol::TMValidation const& m, uint256 const& uid, PublicKey const& validator)
1238{
1239 if (auto const toSkip = app_.getHashRouter().shouldRelay(uid))
1240 {
1241 auto const sm = std::make_shared<Message>(m, protocol::mtVALIDATION, validator);
1242 forEach([&](std::shared_ptr<PeerImp> const& p) {
1243 if (!toSkip->contains(p->id()))
1244 p->send(sm);
1245 });
1246 return *toSkip;
1247 }
1248 return {};
1249}
1250
1253{
1255
1256 if (auto seq = app_.getValidatorManifests().sequence(); seq != manifestListSeq_)
1257 {
1258 // Phase 1: snapshot the cache under its own lock. Do not call
1259 // Validators::listed() here — that takes the validator-list lock, and
1260 // forEachManifest holds the manifest-cache lock, so consulting trust
1261 // inside the callback would invert the lock order used elsewhere
1262 // (see onManifests) and risk deadlock. Capture the manifest hash now,
1263 // while we have the Manifest object, for the suppression key.
1264 struct CachedManifest
1265 {
1266 PublicKey masterKey;
1267 std::string serialized;
1268 uint256 hash;
1269 };
1271 app_.getValidatorManifests().forEachManifest(
1272 [&cached](std::size_t s) { cached.reserve(s); },
1273 [&cached](Manifest const& manifest) {
1274 cached.push_back(
1275 {.masterKey = manifest.masterKey,
1276 .serialized = manifest.serialized,
1277 .hash = manifest.hash()});
1278 });
1279
1280 // Phase 2: no cache lock held, so trust checks are safe. Include every
1281 // trusted manifest, then fill any remaining headroom up to the
1282 // configured untrusted count with gossip. Trusted manifests are never
1283 // dropped; the trusted count only sizes the accepted message.
1286 for (auto const& e : cached)
1287 {
1288 if (app_.getValidators().listed(e.masterKey))
1289 {
1290 selected.push_back(&e);
1291 }
1292 else
1293 {
1294 untrusted.push_back(&e);
1295 }
1296 }
1297
1298 // Cap untrusted only; trusted manifests are all included above.
1299 auto const take =
1300 std::min(untrustedManifestCount(app_.config().maxUntrustedCount), untrusted.size());
1301 selected.insert(selected.end(), untrusted.begin(), untrusted.begin() + take);
1302
1303 // Shuffle the order. Cryptographic randomness is not needed here.
1304 std::shuffle(selected.begin(), selected.end(), defaultPrng());
1305
1306 protocol::TMManifests tm;
1307 auto& hr = app_.getHashRouter();
1308 tm.mutable_list()->Reserve(static_cast<int>(selected.size()));
1309 for (auto const* e : selected)
1310 {
1311 tm.add_list()->set_stobject(e->serialized.data(), e->serialized.size());
1312 hr.addSuppression(e->hash);
1313 }
1314
1315 manifestMessage_.reset();
1316
1317 if (tm.list_size() != 0)
1318 manifestMessage_ = std::make_shared<Message>(tm, protocol::mtMANIFESTS);
1319
1320 manifestListSeq_ = seq;
1321 }
1322
1323 return manifestMessage_;
1324}
1325
1326void
1328 uint256 const& hash,
1330 std::set<Peer::id_t> const& toSkip)
1331{
1332 bool relay = tx.has_value();
1333 if (relay)
1334 {
1335 auto& txn = tx->get();
1336 SerialIter sit(makeSlice(txn.rawtransaction()));
1337 try
1338 {
1339 relay = !isPseudoTx(STTx{sit});
1340 }
1341 catch (std::exception const&)
1342 {
1343 // Could not construct STTx, not relaying
1344 JLOG(journal_.debug()) << "Could not construct STTx: " << hash;
1345 return;
1346 }
1347 }
1348
1349 Overlay::PeerSequence peers = {};
1350 std::size_t total = 0;
1351 std::size_t disabled = 0;
1352 std::size_t enabledInSkip = 0;
1353
1354 if (!relay)
1355 {
1356 if (!app_.config().txReduceRelayEnable)
1357 return;
1358
1359 peers = getActivePeers(toSkip, total, disabled, enabledInSkip);
1360 JLOG(journal_.trace()) << "not relaying tx, total peers " << peers.size();
1361 for (auto const& p : peers)
1362 p->addTxQueue(hash);
1363 return;
1364 }
1365
1366 auto& txn = tx->get();
1367 auto const sm = std::make_shared<Message>(txn, protocol::mtTRANSACTION);
1368 peers = getActivePeers(toSkip, total, disabled, enabledInSkip);
1369 auto const minRelay = app_.config().txReduceRelayMinPeers + disabled;
1370
1371 if (!app_.config().txReduceRelayEnable || total <= minRelay)
1372 {
1373 for (auto const& p : peers)
1374 p->send(sm);
1375 if (app_.config().txReduceRelayEnable || app_.config().txReduceRelayMetrics)
1376 txMetrics_.addMetrics(total, toSkip.size(), 0);
1377 return;
1378 }
1379
1380 // We have more peers than the minimum (disabled + minimum enabled),
1381 // relay to all disabled and some randomly selected enabled that
1382 // do not have the transaction.
1383 auto const enabledTarget = app_.config().txReduceRelayMinPeers +
1384 ((total - minRelay) * app_.config().txRelayPercentage / 100);
1385
1386 txMetrics_.addMetrics(enabledTarget, toSkip.size(), disabled);
1387
1388 if (enabledTarget > enabledInSkip)
1389 std::shuffle(peers.begin(), peers.end(), defaultPrng());
1390
1391 JLOG(journal_.trace()) << "relaying tx, total peers " << peers.size() << " selected "
1392 << enabledTarget << " skip " << toSkip.size() << " disabled "
1393 << disabled;
1394
1395 // count skipped peers with the enabled feature towards the quota
1396 std::uint16_t enabledAndRelayed = enabledInSkip;
1397 for (auto const& p : peers)
1398 {
1399 // always relay to a peer with the disabled feature
1400 if (!p->txReduceRelayEnabled())
1401 {
1402 p->send(sm);
1403 }
1404 else if (enabledAndRelayed < enabledTarget)
1405 {
1406 enabledAndRelayed++;
1407 p->send(sm);
1408 }
1409 else
1410 {
1411 p->addTxQueue(hash);
1412 }
1413 }
1414}
1415
1416//------------------------------------------------------------------------------
1417
1418void
1420{
1422 list_.erase(&child);
1423 if (list_.empty())
1424 cond_.notify_all();
1425}
1426
1427void
1429{
1430 // Calling list_[].second->stop() may cause list_ to be modified
1431 // (OverlayImpl::remove() may be called on this same thread). So
1432 // iterating directly over list_ to call child->stop() could lead to
1433 // undefined behavior.
1434 //
1435 // Therefore we copy all of the weak/shared ptrs out of list_ before we
1436 // start calling stop() on them. That guarantees OverlayImpl::remove()
1437 // won't be called until vector<> children leaves scope.
1439 {
1441 if (!work_)
1442 return;
1443 work_ = std::nullopt;
1444
1445 children.reserve(list_.size());
1446 for (auto const& element : list_)
1447 {
1448 children.emplace_back(element.second.lock());
1449 }
1450 } // lock released
1451
1452 for (auto const& child : children)
1453 {
1454 if (child != nullptr)
1455 child->stop();
1456 }
1457}
1458
1459void
1461{
1462 auto const result = peerFinder_->autoconnect();
1463 for (auto const& addr : result)
1464 connect(addr);
1465}
1466
1467void
1469{
1470 auto const result = peerFinder_->buildEndpointsForPeers();
1471 for (auto const& e : result)
1472 {
1474 {
1476 auto const iter = peers_.find(e.first);
1477 if (iter != peers_.end())
1478 peer = iter->second.lock();
1479 }
1480 if (peer)
1481 peer->sendEndpoints(e.second.begin(), e.second.end());
1482 }
1483}
1484
1485void
1487{
1488 forEach([](auto const& p) {
1489 if (p->txReduceRelayEnabled())
1490 p->sendTxQueue();
1491 });
1492}
1493
1495makeSquelchMessage(PublicKey const& validator, bool squelch, uint32_t squelchDuration)
1496{
1497 protocol::TMSquelch m;
1498 m.set_squelch(squelch);
1499 m.set_validatorpubkey(validator.data(), validator.size());
1500 if (squelch)
1501 m.set_squelchduration(squelchDuration);
1502 return std::make_shared<Message>(m, protocol::mtSQUELCH);
1503}
1504
1505void
1507{
1508 if (auto peer = findPeerByShortID(id); peer)
1509 {
1510 // optimize - multiple message with different
1511 // validator might be sent to the same peer
1512 peer->send(makeSquelchMessage(validator, false, 0));
1513 }
1514}
1515
1516void
1517OverlayImpl::squelch(PublicKey const& validator, Peer::id_t id, uint32_t squelchDuration) const
1518{
1519 if (auto peer = findPeerByShortID(id); peer)
1520 {
1521 peer->send(makeSquelchMessage(validator, true, squelchDuration));
1522 }
1523}
1524
1525void
1527 uint256 const& key,
1528 PublicKey const& validator,
1529 std::set<Peer::id_t>&& peers,
1530 protocol::MessageType type)
1531{
1532 if (!slots_.baseSquelchReady())
1533 return;
1534
1535 if (!strand_.running_in_this_thread())
1536 {
1537 post(
1538 strand_,
1539 // Must capture copies of reference parameters (i.e. key, validator)
1540 [this, key = key, validator = validator, peers = std::move(peers), type]() mutable {
1541 updateSlotAndSquelch(key, validator, std::move(peers), type);
1542 });
1543
1544 return;
1545 }
1546
1547 for (auto id : peers)
1548 {
1549 slots_.updateSlotAndSquelch(key, validator, id, type, [&]() {
1551 });
1552 }
1553}
1554
1555void
1557 uint256 const& key,
1558 PublicKey const& validator,
1559 Peer::id_t peer,
1560 protocol::MessageType type)
1561{
1562 if (!slots_.baseSquelchReady())
1563 return;
1564
1565 if (!strand_.running_in_this_thread())
1566 {
1567 {
1568 post(
1569 strand_,
1570 // Must capture copies of reference parameters (i.e. key, validator)
1571 [this, key = key, validator = validator, peer, type]() {
1572 updateSlotAndSquelch(key, validator, peer, type);
1573 });
1574 }
1575 return;
1576 }
1577
1578 slots_.updateSlotAndSquelch(key, validator, peer, type, [&]() {
1580 });
1581}
1582
1583void
1585{
1586 if (!strand_.running_in_this_thread())
1587 {
1588 post(strand_, [this, id] { deletePeer(id); });
1589 return;
1590 }
1591
1592 slots_.deletePeer(id, true);
1593}
1594
1595void
1597{
1598 if (!strand_.running_in_this_thread())
1599 {
1600 post(strand_, [this] { deleteIdlePeers(); });
1601 return;
1602 }
1603
1604 slots_.deleteIdlePeers();
1605}
1606
1607//------------------------------------------------------------------------------
1608
1611{
1612 Overlay::Setup setup;
1613
1614 {
1615 auto const& section = config.section(Sections::kOverlay);
1616 setup.context = makeSslContext("");
1617
1618 set(setup.ipLimit, "ip_limit", section);
1619 if (setup.ipLimit < 0)
1620 Throw<std::runtime_error>("Configured IP limit is invalid");
1621
1622 std::string ip;
1623 set(ip, "public_ip", section);
1624 if (!ip.empty())
1625 {
1626 boost::system::error_code ec;
1627 setup.publicIp = boost::asio::ip::make_address(ip, ec);
1628 if (ec || !beast::ip::isPublic(setup.publicIp))
1629 Throw<std::runtime_error>("Configured public IP is invalid");
1630 }
1631
1632 set(setup.verifyEndpoints, true, "verify_endpoints", section);
1633 if (!setup.verifyEndpoints)
1634 {
1635 JLOG(j.warn()) << "Endpoint verification is disabled. This is a "
1636 "security risk and should only be used for "
1637 "testing.";
1638 }
1639 }
1640
1641 {
1642 auto const& section = config.section(Sections::kCrawl);
1643 auto const& values = section.values();
1644
1645 if (values.size() > 1)
1646 {
1647 Throw<std::runtime_error>("Configured [crawl] section is invalid, too many values");
1648 }
1649
1650 bool crawlEnabled = true;
1651
1652 // Only allow "0|1" as a value
1653 if (values.size() == 1)
1654 {
1655 try
1656 {
1657 crawlEnabled = boost::lexical_cast<bool>(values.front());
1658 }
1659 catch (boost::bad_lexical_cast const&)
1660 {
1662 "Configured [crawl] section has invalid value: " + values.front());
1663 }
1664 }
1665
1666 if (crawlEnabled)
1667 {
1668 if (get<bool>(section, Keys::kOverlay, true))
1669 {
1671 }
1672 if (get<bool>(section, Keys::kServer, true))
1673 {
1675 }
1676 if (get<bool>(section, Keys::kCounts, false))
1677 {
1679 }
1680 if (get<bool>(section, Keys::kUnl, true))
1681 {
1683 }
1684 }
1685 }
1686 {
1687 auto const& section = config.section(Sections::kVl);
1688
1689 set(setup.vlEnabled, "enabled", section);
1690 }
1691
1692 try
1693 {
1694 auto id = config.legacy(Sections::kNetworkId);
1695
1696 if (!id.empty())
1697 {
1698 if (id == "main")
1699 id = "0";
1700
1701 if (id == "testnet")
1702 id = "1";
1703
1704 if (id == "devnet")
1705 id = "2";
1706
1708 }
1709 }
1710 catch (...)
1711 {
1713 "Configured [network_id] section is invalid: must be a number "
1714 "or one of the strings 'main', 'testnet' or 'devnet'.");
1715 }
1716
1717 return setup;
1718}
1719
1722 Application& app,
1723 Overlay::Setup const& setup,
1724 ServerHandler& serverHandler,
1725 resource::Manager& resourceManager,
1726 Resolver& resolver,
1727 boost::asio::io_context& ioContext,
1728 BasicConfig const& config,
1729 beast::insight::Collector::ptr const& collector)
1730{
1732 app, setup, serverHandler, resourceManager, resolver, ioContext, config, collector);
1733}
1734
1735} // namespace xrpl
T begin(T... args)
A generic endpoint for log messages.
Definition Journal.h:44
Stream debug() const
Definition Journal.h:344
Stream warn() const
Definition Journal.h:356
std::string const & name() const
Returns the name of this source.
void add(Source &source)
Add a child source.
Wraps a Journal::Sink to prefix its output with a string.
Definition WrappedSink.h:19
std::shared_ptr< Collector > ptr
Definition Collector.h:29
A version-independent IP address and port combination.
Definition IPEndpoint.h:24
Represents a JSON value.
Definition json_value.h:117
Value removeMember(char const *key)
Remove and return the named member.
Value & append(Value const &value)
Append value to array at the end.
bool isMember(char const *key) const
Return true if the object has a member named key.
Holds unparsed configuration information.
void legacy(std::string const &section, std::string value)
Set a value that is not a key/value pair.
Section & section(std::string const &name)
Returns the section with the given name.
Child(OverlayImpl &overlay)
void deletePeer(Peer::id_t id)
Called when the peer is deleted.
boost::asio::io_context & ioContext_
resource::Manager & resourceManager()
bool processRequest(http_request_type const &req, Handoff &handoff)
Handles non-peer protocol requests.
json::Value getOverlayInfo() const
Returns information about peers on the overlay network.
boost::asio::ip::address address_type
Definition OverlayImpl.h:88
static bool isPeerUpgrade(http_request_type const &request)
Resolver & resolver_
void addActive(std::shared_ptr< PeerImp > const &peer)
boost::system::error_code error_code
Definition OverlayImpl.h:90
bool processCrawl(http_request_type const &req, Handoff &handoff)
Handles crawl requests.
void broadcast(protocol::TMProposeSet const &m) override
Broadcast a proposal.
bool processHealth(http_request_type const &req, Handoff &handoff)
Handles health requests.
void activate(std::shared_ptr< PeerImp > const &peer)
Called when a peer has connected successfully This is called after the peer handshake has been comple...
std::optional< boost::asio::executor_work_guard< boost::asio::io_context::executor_type > > work_
void connect(beast::ip::Endpoint const &remoteEndpoint) override
Establish a peer connection to the specified endpoint.
peer_finder::Manager & peerFinder()
void remove(std::shared_ptr< peer_finder::Slot > const &slot)
std::set< Peer::id_t > relay(protocol::TMProposeSet const &m, uint256 const &uid, PublicKey const &validator) override
Relay a proposal.
void stop() override
std::size_t size() const override
The number of active peers on the network Active peers are only those peers that have completed the h...
static bool isUpgrade(boost::beast::http::header< true, Fields > const &req)
ServerHandler & serverHandler_
void onManifests(std::shared_ptr< protocol::TMManifests > const &m, std::shared_ptr< PeerImp > const &from)
static std::shared_ptr< Writer > makeErrorResponse(std::shared_ptr< peer_finder::Slot > const &slot, http_request_type const &request, address_type remoteAddress, std::string const &msg)
std::shared_ptr< Writer > makeRedirectResponse(std::shared_ptr< peer_finder::Slot > const &slot, http_request_type const &request, address_type remoteAddress)
Handoff onHandoff(std::unique_ptr< stream_type > &&bundle, http_request_type &&request, endpoint_type remoteEndpoint) override
Conditionally accept an incoming HTTP request.
peer_finder::StoreSqdb store_
reduce_relay::Slots< UptimeClock > slots_
hash_map< Peer::id_t, std::weak_ptr< PeerImp > > ids_
void deleteIdlePeers()
Check if peers stopped relaying messages and if slots stopped receiving messages from the validator.
OverlayImpl(Application &app, Setup setup, ServerHandler &serverHandler, resource::Manager &resourceManager, Resolver &resolver, boost::asio::io_context &ioContext, BasicConfig const &config, beast::insight::Collector::ptr const &collector)
void squelch(PublicKey const &validator, Peer::id_t const id, std::uint32_t squelchDuration) const override
Squelch handler.
void reportInboundTraffic(TrafficCount::Category cat, int bytes)
std::shared_ptr< Message > manifestMessage_
void sendTxQueue() const
Send once a second transactions' hashes aggregated by peers.
std::optional< std::uint32_t > manifestListSeq_
std::unique_ptr< peer_finder::Manager > peerFinder_
void onWrite(beast::PropertyStream::Map &stream) override
Subclass override.
resource::Manager & resourceManager_
Application & app_
std::atomic< Peer::id_t > nextId_
json::Value getServerCounts()
Returns information about the local server's performance counters.
std::recursive_mutex mutex_
beast::Journal const journal_
json::Value json() override
Return diagnostics on the status of all peers.
boost::asio::ip::tcp::endpoint endpoint_type
Definition OverlayImpl.h:89
void forEach(UnaryFunc &&f) const
void onPeerDeactivate(Peer::id_t id)
std::mutex manifestLock_
boost::asio::strand< boost::asio::io_context::executor_type > strand_
static std::string makePrefix(std::uint32_t id)
Setup const & setup() const
metrics::TxMetrics txMetrics_
boost::container::flat_map< Child *, std::weak_ptr< Child > > list_
int limit() override
Returns the maximum number of peers we are configured to allow.
std::condition_variable_any cond_
json::Value getUnlInfo()
Returns information about the local server's UNL.
std::shared_ptr< Message > getManifestsMessage()
std::shared_ptr< Peer > findPeerByPublicKey(PublicKey const &pubKey) override
Returns the peer with the matching public key, or null.
TrafficCount traffic_
hash_map< std::shared_ptr< peer_finder::Slot >, std::weak_ptr< PeerImp > > peers_
bool processValidatorList(http_request_type const &req, Handoff &handoff)
Handles validator list requests.
void checkTracking(std::uint32_t) override
Calls the checkTracking function on each peer.
json::Value getServerInfo()
Returns information about the local server.
void updateSlotAndSquelch(uint256 const &key, PublicKey const &validator, std::set< Peer::id_t > &&peers, protocol::MessageType type)
Updates message count for validator/peer.
void reportOutboundTraffic(TrafficCount::Category cat, int bytes)
std::shared_ptr< Peer > findPeerByShortID(Peer::id_t const &id) const override
Returns the peer with the matching short id, or null.
void start() override
PeerSequence getActivePeers() const override
Returns a sequence representing the current list of peers.
void unsquelch(PublicKey const &validator, Peer::id_t id) const override
Unsquelch handler.
std::vector< std::shared_ptr< Peer > > PeerSequence
Definition Overlay.h:66
std::uint32_t id_t
Uniquely identifies a peer.
A public key.
Definition PublicKey.h:53
std::vector< std::string > const & values() const
Returns all the values in the section.
Definition BasicConfig.h:69
virtual std::pair< std::shared_ptr< Slot >, Result > newOutboundSlot(beast::ip::Endpoint const &remoteEndpoint)=0
Create a new outbound slot with the specified remote endpoint.
Tracks load and resource consumption.
virtual Consumer newOutboundEndpoint(beast::ip::Endpoint const &address)=0
Create a new endpoint keyed by outbound IP address and port.
T contains(T... args)
T duration_cast(T... args)
T emplace_back(T... args)
T emplace(T... args)
T empty(T... args)
T end(T... args)
T find_if(T... args)
T get(T... args)
T insert(T... args)
T lock(T... args)
T make_shared(T... args)
T make_tuple(T... args)
T make_unique(T... args)
T max(T... args)
T min(T... args)
bool isPublic(Address const &addr)
Returns true if the address is a public routable address.
Definition IPAddress.h:71
bool isKeepAlive(boost::beast::http::message< IsRequest, Body, Fields > const &m)
Definition rfc2616.h:366
Result splitCommas(FwdIt first, FwdIt last)
Definition rfc2616.h:182
constexpr Out lexicalCastThrow(In in)
Convert from one type to another, throw on error.
JSON (JavaScript Object Notation).
Definition json_errors.h:5
@ Array
array value (ordered list)
Definition json_value.h:28
@ Object
object value (collection of name/value pairs).
Definition json_value.h:29
@ Null
'null' value
Definition json_value.h:22
STL namespace.
std::string const & getFullVersionString()
Full server version string.
Definition BuildInfo.cpp:82
static constexpr auto kDisabled
static constexpr auto kOverlay
static constexpr auto kUnl
static constexpr auto kServerCounts
static constexpr auto kServerInfo
Config makeConfig(xrpl::Config const &cfg, std::uint16_t port, bool validationPublicKey, int ipLimit, bool verifyEndpoints)
Charge const kFeeMalformedRequest
Schedule of fees charged for imposing load on the server.
static constexpr auto kCheckIdlePeers
How often we check for idle peers (seconds).
Use hash_* containers for keys that do not need a cryptographically secure hashing algorithm.
Definition algorithm.h:5
std::vector< ProtocolVersion > parseProtocolVersions(boost::beast::string_view const &value)
Parse a set of protocol versions.
bool set(T &target, std::string const &name, Section const &section)
Set a value from a configuration Section If the named value is not found or doesn't parse as a T,...
std::optional< uint256 > makeSharedValue(stream_type &ssl, beast::Journal journal)
Computes a shared value based on the SSL connection state.
Stopwatch & stopwatch()
Returns an instance of a wall clock.
Definition chrono.h:101
T get(Section const &section, std::string const &name, T const &defaultValue=T{})
Retrieve a key/value pair from a section.
std::string strHex(FwdIt begin, FwdIt end)
Definition strHex.h:13
std::optional< ProtocolVersion > negotiateProtocolVersion(std::vector< ProtocolVersion > const &versions)
Given a list of supported protocol versions, choose the one we prefer.
std::string to_string(BaseUInt< Bits, Tag > const &a)
Definition base_uint.h:651
void addValidatorManifest(soci::session &session, std::string const &serialized)
addValidatorManifest Saves the manifest of a validator to the database.
Definition Wallet.cpp:137
std::optional< Manifest > deserializeManifest(Slice s, beast::Journal journal)
Constructs Manifest from serialized string.
@ Uncapped
Bypasses the cap (listed/trusted or config manifests).
Definition Manifest.h:365
@ Capped
Subject to the untrusted cap (unlisted peer gossip).
Definition Manifest.h:364
Slice makeSlice(std::array< T, N > const &a)
Definition Slice.h:228
std::shared_ptr< boost::asio::ssl::context > makeSslContext(std::string const &cipherList)
Create a self-signed SSL context that allows anonymous Diffie Hellman.
std::string base64Encode(std::uint8_t const *data, std::size_t len)
std::unique_ptr< Overlay > makeOverlay(Application &app, Overlay::Setup const &setup, ServerHandler &serverHandler, resource::Manager &resourceManager, Resolver &resolver, boost::asio::io_context &ioContext, BasicConfig const &config, beast::insight::Collector::ptr const &collector)
Creates the implementation of Overlay.
constexpr Number squelch(Number const &x, Number const &limit) noexcept
Definition Number.h:907
beast::xor_shift_engine & defaultPrng()
Return the default random engine.
std::shared_ptr< Message > makeSquelchMessage(PublicKey const &validator, bool squelch, uint32_t squelchDuration)
boost::beast::http::request< boost::beast::http::dynamic_body > http_request_type
Definition Handoff.h:12
constexpr std::size_t untrustedManifestCount(std::optional< std::size_t > const &configured)
Number of untrusted manifests to store in cache and allowed in one Manifest message.
Definition Manifest.h:246
Overlay::Setup setupOverlay(BasicConfig const &config, beast::Journal j)
json::Value getCountsJson(Application &app, int minObjectCount)
Definition GetCounts.cpp:46
bool isPseudoTx(STObject const &tx)
Check whether a transaction is a pseudo-transaction.
Definition STTx.cpp:886
BaseUInt< 256 > uint256
Definition base_uint.h:580
PublicKey verifyHandshake(boost::beast::http::fields const &headers, xrpl::uint256 const &sharedValue, std::optional< std::uint32_t > networkID, beast::ip::Address publicIp, beast::ip::Address remote, Application &app)
Validate header fields necessary for upgrading the link to the peer protocol.
XRPL_NO_SANITIZE_ADDRESS void Throw(Args &&... args)
Definition contract.h:52
@ Accepted
Manifest is valid.
Definition Manifest.h:321
T piecewise_construct
T push_back(T... args)
T shuffle(T... args)
T reserve(T... args)
T setfill(T... args)
T setw(T... args)
T size(T... args)
T str(T... args)
static boost::asio::ip::tcp::endpoint toAsioEndpoint(ip::Endpoint const &address)
static ip::Endpoint fromAsio(boost::asio::ip::address const &address)
Used to indicate the result of a server connection handoff.
Definition Handoff.h:20
bool keepAlive
Definition Handoff.h:26
std::shared_ptr< Writer > response
Definition Handoff.h:29
static constexpr auto kUnl
Definition Constants.h:175
static constexpr auto kCounts
Definition Constants.h:103
static constexpr auto kOverlay
Definition Constants.h:139
static constexpr auto kServer
Definition Constants.h:156
void onTimer(error_code ec)
boost::asio::basic_waitable_timer< clock_type > timer
Definition OverlayImpl.h:94
Timer(OverlayImpl &overlay)
std::uint32_t crawlOptions
Definition Overlay.h:60
std::optional< std::uint32_t > networkID
Definition Overlay.h:61
beast::ip::Address publicIp
Definition Overlay.h:58
std::shared_ptr< boost::asio::ssl::context > context
Definition Overlay.h:57
static constexpr auto kOverlay
Definition Constants.h:35
static constexpr auto kVl
Definition Constants.h:77
static constexpr auto kCrawl
Definition Constants.h:12
static constexpr auto kNetworkId
Definition Constants.h:30
PeerFinder configuration settings.
T substr(T... args)
T to_string(T... args)
T what(T... args)