ARCHIVE/xahaud
3
0
Fork 0
mirror of https://github.com/Xahau/xahaud.git synced 2025-12-06 17:27:52 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
e442a2846d7facc21a61ff5ad33076b19a038038
xahaud/src/ripple/overlay/impl/PeerImp.cpp
Vinnie Falco e442a2846d Overlay improvements and bug fixes: 
PeerImp::detach had a default argument graceful=true which did not
correctly close the socket and cause the Overlay to often hang on exit.
The logging for Overlay and Peers has been reworked. All the socket activity
is logged to Peers while protocol activity goes to Protocol. Every log line
is prefixed by a small integer ID unique to the connection.
* Removed graceful PeerImp::detach option
* Peer and Protocol log message handle respective types of logging
* Log messages prefixed with peer unique integer
* Prevent call to timer ancel from throwing an exception
2014-11-08 14:39:46 -08:00

2709 lines
77 KiB
C++
Raw Blame History

//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
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 OR ANY 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 <ripple/basics/StringUtilities.h>
#include <ripple/basics/UptimeTimer.h>
#include <ripple/core/JobQueue.h>
#include <ripple/json/json_reader.h>
#include <ripple/overlay/impl/PeerImp.h>
#include <ripple/overlay/impl/Tuning.h>
#include <ripple/resource/Fees.h>
#include <ripple/server/ServerHandler.h>
#include <ripple/protocol/BuildInfo.h>
#include <beast/streams/debug_ostream.h>
#include <functional>
#include <beast/cxx14/memory.h> // <memory>
#include <sstream>
namespace ripple {
PeerImp::PeerImp (std::unique_ptr<beast::asio::ssl_bundle>&& ssl_bundle,
beast::http::message&& request, beast::IP::Endpoint remoteAddress,
OverlayImpl& overlay, Resource::Manager& resourceManager,
PeerFinder::Manager& peerFinder, PeerFinder::Slot::ptr const& slot)
: Child (overlay)
, sink_(deprecatedLogs().journal("Peer"))
, p_sink_(deprecatedLogs().journal("Protocol"))
, journal_ (sink_)
, p_journal_(p_sink_)
, ssl_bundle_(std::move(ssl_bundle))
, socket_ (ssl_bundle_->socket)
, stream_ (ssl_bundle_->stream)
, strand_ (socket_.get_io_service())
, timer_ (socket_.get_io_service())
, remote_address_ (remoteAddress)
, resourceManager_ (resourceManager)
, peerFinder_ (peerFinder)
, overlay_ (overlay)
, m_inbound (true)
, state_ (stateConnected)
, slot_ (slot)
, message_stream_(*this)
{
setPrefix();
}
PeerImp::PeerImp (beast::IP::Endpoint remoteAddress,
boost::asio::io_service& io_service, OverlayImpl& overlay,
Resource::Manager& resourceManager, PeerFinder::Manager& peerFinder,
PeerFinder::Slot::ptr const& slot,
std::shared_ptr<boost::asio::ssl::context> const& context)
: Child (overlay)
, sink_(deprecatedLogs().journal("Peer"))
, p_sink_(deprecatedLogs().journal("Protocol"))
, journal_ (sink_)
, p_journal_(p_sink_)
, ssl_bundle_(std::make_unique<beast::asio::ssl_bundle>(
context, io_service))
, socket_ (ssl_bundle_->socket)
, stream_ (ssl_bundle_->stream)
, strand_ (socket_.get_io_service())
, timer_ (socket_.get_io_service())
, remote_address_ (remoteAddress)
, resourceManager_ (resourceManager)
, peerFinder_ (peerFinder)
, overlay_ (overlay)
, m_inbound (false)
, state_ (stateConnecting)
, slot_ (slot)
, message_stream_(*this)
{
setPrefix();
}
PeerImp::~PeerImp ()
{
overlay_.remove (slot_);
}
void
PeerImp::start()
{
boost::system::error_code ec;
timer_.expires_from_now (nodeVerifySeconds, ec);
timer_.async_wait (strand_.wrap (std::bind (&PeerImp::handleVerifyTimer,
shared_from_this (), beast::asio::placeholders::error)));
if (ec)
{
journal_.error << "Failed to set verify timer.";
return detach ("start");
}
if (m_inbound)
do_accept();
else
do_connect();
}
void
PeerImp::close()
{
detach("close");
}
//------------------------------------------------------------------------------
void
PeerImp::send (Message::pointer const& m)
{
// VFALCO NOTE why call this with null?
if (! m)
return;
if (! strand_.running_in_this_thread())
{
strand_.post (std::bind (&PeerImp::send, shared_from_this(), m));
return;
}
if (send_packet_)
send_queue_.push_back (m);
else
sendForce (m);
}
beast::IP::Endpoint
PeerImp::getRemoteAddress() const
{
return remote_address_;
}
void
PeerImp::charge (Resource::Charge const& fee)
{
if ((usage_.charge (fee) == Resource::drop) && usage_.disconnect ())
detach ("resource");
}
//------------------------------------------------------------------------------
Peer::ShortId
PeerImp::getShortId () const
{
return shortId_;
}
RippleAddress const&
PeerImp::getNodePublic () const
{
return publicKey_;
}
Json::Value
PeerImp::json()
{
Json::Value ret (Json::objectValue);
ret["public_key"] = publicKey_.ToString ();
ret["address"] = remote_address_.to_string();
if (m_inbound)
ret["inbound"] = true;
if (clusterNode_)
{
ret["cluster"] = true;
if (!name_.empty ())
ret["name"] = name_;
}
if (hello_.has_fullversion ())
ret["version"] = hello_.fullversion ();
if (hello_.has_protoversion ())
{
auto protocol = BuildInfo::make_protocol (hello_.protoversion ());
if (protocol != BuildInfo::getCurrentProtocol())
ret["protocol"] = to_string (protocol);
}
std::uint32_t minSeq, maxSeq;
ledgerRange(minSeq, maxSeq);
if ((minSeq != 0) || (maxSeq != 0))
ret["complete_ledgers"] = boost::lexical_cast<std::string>(minSeq) +
" - " + boost::lexical_cast<std::string>(maxSeq);
if (closedLedgerHash_ != zero)
ret["ledger"] = to_string (closedLedgerHash_);
if (last_status_.has_newstatus ())
{
switch (last_status_.newstatus ())
{
case protocol::nsCONNECTING:
ret["status"] = "connecting";
break;
case protocol::nsCONNECTED:
ret["status"] = "connected";
break;
case protocol::nsMONITORING:
ret["status"] = "monitoring";
break;
case protocol::nsVALIDATING:
ret["status"] = "validating";
break;
case protocol::nsSHUTTING:
ret["status"] = "shutting";
break;
default:
// FIXME: do we really want this?
p_journal_.warning <<
"Unknown status: " << last_status_.newstatus ();
}
}
return ret;
}
bool
PeerImp::isInCluster () const
{
return clusterNode_;
}
std::string const&
PeerImp::getClusterNodeName() const
{
return name_;
}
//------------------------------------------------------------------------------
uint256 const&
PeerImp::getClosedLedgerHash () const
{
return closedLedgerHash_;
}
bool
PeerImp::hasLedger (uint256 const& hash, std::uint32_t seq) const
{
std::lock_guard<std::mutex> sl(recentLock_);
if ((seq != 0) && (seq >= minLedger_) && (seq <= maxLedger_))
return true;
return std::find (recentLedgers_.begin(),
recentLedgers_.end(), hash) != recentLedgers_.end();
}
void
PeerImp::ledgerRange (std::uint32_t& minSeq,
std::uint32_t& maxSeq) const
{
std::lock_guard<std::mutex> sl(recentLock_);
minSeq = minLedger_;
maxSeq = maxLedger_;
}
bool
PeerImp::hasTxSet (uint256 const& hash) const
{
std::lock_guard<std::mutex> sl(recentLock_);
return std::find (recentTxSets_.begin(),
recentTxSets_.end(), hash) != recentTxSets_.end();
}
void
PeerImp::cycleStatus ()
{
previousLedgerHash_ = closedLedgerHash_;
closedLedgerHash_.zero ();
}
bool
PeerImp::supportsVersion (int version)
{
return hello_.has_protoversion () && (hello_.protoversion () >= version);
}
bool
PeerImp::hasRange (std::uint32_t uMin, std::uint32_t uMax)
{
return (uMin >= minLedger_) && (uMax <= maxLedger_);
}
//------------------------------------------------------------------------------
// Puts a string at the beginning of each string logged to the journal
void
PeerImp::setPrefix()
{
static std::atomic<std::size_t> id;
std::stringstream ss;
ss << "[" << std::setfill('0') << std::setw(3) << ++id << "] ";
sink_.prefix(ss.str());
p_sink_.prefix(ss.str());
}
/* Completion handlers for client role.
Logic steps:
1. Establish outgoing connection
2. Perform SSL handshake
3. Send HTTP request
4. Receive HTTP response
5. Enter protocol loop
*/
void PeerImp::do_connect ()
{
journal_.info << "Connecting to " << remote_address_;
usage_ = resourceManager_.newOutboundEndpoint (remote_address_);
if (usage_.disconnect ())
return detach ("do_connect");
stream_.next_layer().async_connect (
beast::IPAddressConversion::to_asio_endpoint (remote_address_),
strand_.wrap (std::bind (&PeerImp::on_connect,
shared_from_this (), beast::asio::placeholders::error)));
}
void
PeerImp::on_connect (error_code ec)
{
if (detaching_ || ec == boost::asio::error::operation_aborted)
return;
socket_type::endpoint_type local_endpoint;
if (! ec)
local_endpoint = stream_.next_layer().local_endpoint (ec);
if (ec)
{
journal_.error <<
"Connect to " << remote_address_ <<
" failed: " << ec.message();
return detach ("hc");
}
assert (state_ == stateConnecting);
state_ = stateConnected;
if (! peerFinder_.connected (slot_,
beast::IPAddressConversion::from_asio (local_endpoint)))
return detach("dup");
stream_.set_verify_mode (boost::asio::ssl::verify_none);
stream_.async_handshake (
boost::asio::ssl::stream_base::client,
strand_.wrap (std::bind (&PeerImp::on_connect_ssl,
std::static_pointer_cast <PeerImp> (shared_from_this ()),
beast::asio::placeholders::error)));
}
beast::http::message
PeerImp::make_request()
{
assert (! m_inbound);
beast::http::message m;
m.method (beast::http::method_t::http_get);
m.url ("/");
m.version (1, 1);
m.headers.append ("User-Agent", BuildInfo::getFullVersionString());
//m.headers.append ("Local-Address", stream_.
m.headers.append ("Remote-Address", remote_address_.to_string());
m.headers.append ("Upgrade",
std::string("Ripple/") + to_string (BuildInfo::getCurrentProtocol()));
m.headers.append ("Connection", "Upgrade");
m.headers.append ("Connect-As", "Leaf, Peer");
m.headers.append ("Accept-Encoding", "identity, snappy");
//m.headers.append ("X-Try-IPs", "192.168.0.1:51234");
//m.headers.append ("X-Try-IPs", "208.239.114.74:51234");
//m.headers.append ("A", "BC");
//m.headers.append ("Content-Length", "0");
return m;
}
void
PeerImp::on_connect_ssl (error_code ec)
{
if (detaching_ || ec == boost::asio::error::operation_aborted)
return;
if (ec)
{
journal_.info <<
"on_connect_ssl: " << ec.message();
detach("on_connect_ssl");
return;
}
if (overlay_.setup().http_handshake)
{
beast::http::message req (make_request());
beast::http::write (write_buffer_, req);
on_write_http_request (error_code(), 0);
}
else
{
do_protocol_start();
}
}
// Called repeatedly with the http request data
void
PeerImp::on_write_http_request (error_code ec, std::size_t bytes_transferred)
{
if (detaching_ || ec == boost::asio::error::operation_aborted)
return;
if (ec)
{
journal_.info <<
"on_write_http_request: " << ec.message();
detach("on_write_http_request");
return;
}
write_buffer_.consume (bytes_transferred);
if (write_buffer_.size() == 0)
{
// done sending request, now read the response
http_message_ = boost::in_place ();
http_body_.clear();
http_parser_ = boost::in_place (std::ref(*http_message_),
std::ref(http_body_), false);
on_read_http_response (error_code(), 0);
return;
}
stream_.async_write_some (write_buffer_.data(),
strand_.wrap (std::bind (&PeerImp::on_write_http_request,
shared_from_this(), beast::asio::placeholders::error,
beast::asio::placeholders::bytes_transferred)));
}
template <class = void>
boost::asio::ip::tcp::endpoint
parse_endpoint (std::string const& s, boost::system::error_code& ec)
{
beast::IP::Endpoint bep;
std::istringstream is(s);
is >> bep;
if (is.fail())
{
ec = boost::system::errc::make_error_code(
boost::system::errc::invalid_argument);
return boost::asio::ip::tcp::endpoint{};
}
return beast::IPAddressConversion::to_asio_endpoint(bep);
}
template <class Streambuf>
void
PeerImp::processResponse (beast::http::message const& m,
Streambuf const& body)
{
if (http_message_->status() == 503)
{
Json::Value json;
Json::Reader r;
auto const success = r.parse(to_string(body), json);
if (success)
{
if (json.isObject() && json.isMember("peer-ips"))
{
Json::Value const& ips = json["peer-ips"];
if (ips.isArray())
{
std::vector<boost::asio::ip::tcp::endpoint> eps;
eps.reserve(ips.size());
for (auto const& v : ips)
{
if (v.isString())
{
error_code ec;
auto const ep = parse_endpoint(v.asString(), ec);
if (!ec)
eps.push_back(ep);
}
}
peerFinder_.onRedirects(beast::IPAddressConversion::
to_asio_endpoint(remote_address_), eps);
}
}
}
}
if (http_message_->status() != 200)
{
if (journal_.info) journal_.info <<
"HTTP Response: " << m.status() << " " << m.reason();
detach("processResponse");
return;
}
do_protocol_start();
}
// Called repeatedly with the http response data
void
PeerImp::on_read_http_response (error_code ec, std::size_t bytes_transferred)
{
if (detaching_ || ec == boost::asio::error::operation_aborted)
return;
if (! ec)
{
read_buffer_.commit (bytes_transferred);
std::size_t bytes_consumed;
std::tie (ec, bytes_consumed) = http_parser_->write (
read_buffer_.data());
if (! ec)
{
read_buffer_.consume (bytes_consumed);
if (http_parser_->complete())
{
return processResponse(*http_message_, http_body_);
}
}
}
if (ec == boost::asio::error::eof)
{
// remote closed their end
// VFALCO TODO Clean up the shutdown of the socket
}
if (ec)
{
journal_.info <<
"on_read_response: " << ec.message();
detach("on_read_response");
return;
}
stream_.async_read_some (read_buffer_.prepare (Tuning::readBufferBytes),
strand_.wrap (std::bind (&PeerImp::on_read_http_response,
shared_from_this(), beast::asio::placeholders::error,
beast::asio::placeholders::bytes_transferred)));
}
//------------------------------------------------------------------------------
/* Completion handlers for server role.
Logic steps:
1. Perform SSL handshake
2. Detect HTTP request or protocol TMHello
3. If HTTP request received, send HTTP response
4. Enter protocol loop
*/
void PeerImp::do_accept()
{
journal_.info << "Accepted " << remote_address_;
usage_ = resourceManager_.newInboundEndpoint (remote_address_);
if (usage_.disconnect ())
{
detach ("do_accept");
return;
}
// VFALCO Hack to receive a legacy protocol connection
// from the HTTP server
if (read_buffer_.size() > 0)
return do_protocol_start();
stream_.set_verify_mode (boost::asio::ssl::verify_none);
stream_.async_handshake (boost::asio::ssl::stream_base::server,
strand_.wrap (std::bind (&PeerImp::on_accept_ssl,
std::static_pointer_cast <PeerImp> (shared_from_this ()),
beast::asio::placeholders::error)));
}
void
PeerImp::on_accept_ssl (error_code ec)
{
if (detaching_ || ec == boost::asio::error::operation_aborted)
return;
if (ec)
{
journal_.info <<
"on_accept_ssl: " << ec.message();
detach("on_accept_ssl");
return;
}
#if 1
on_read_http_detect (error_code(), 0);
#else
do_protocol_start();
#endif
}
// Called repeatedly with the initial bytes received on the connection
void
PeerImp::on_read_http_detect (error_code ec, std::size_t bytes_transferred)
{
if (detaching_ || ec == boost::asio::error::operation_aborted)
return;
if (ec)
{
journal_.info <<
"on_read_detect: " << ec.message();
detach("on_read_detect");
return;
}
read_buffer_.commit (bytes_transferred);
peer_protocol_detector detector;
boost::tribool const is_peer_protocol (detector (read_buffer_.data()));
if (is_peer_protocol)
{
do_protocol_start();
return;
}
else if (! is_peer_protocol)
{
http_message_ = boost::in_place ();
http_body_.clear();
http_parser_ = boost::in_place (std::ref(*http_message_),
std::ref(http_body_), true);
on_read_http_request (error_code(), 0);
return;
}
stream_.async_read_some (read_buffer_.prepare (Tuning::readBufferBytes),
strand_.wrap (std::bind (&PeerImp::on_read_http_detect,
shared_from_this(), beast::asio::placeholders::error,
beast::asio::placeholders::bytes_transferred)));
}
// Called repeatedly with the http request data
void
PeerImp::on_read_http_request (error_code ec, std::size_t bytes_transferred)
{
if (detaching_ || ec == boost::asio::error::operation_aborted)
return;
if (! ec)
{
read_buffer_.commit (bytes_transferred);
std::size_t bytes_consumed;
std::tie (ec, bytes_consumed) = http_parser_->write (
read_buffer_.data());
if (! ec)
{
read_buffer_.consume (bytes_consumed);
if (http_parser_->complete())
{
//
// TODO Apply headers to connection state.
//
if (http_message_->upgrade())
{
std::stringstream ss;
ss <<
"HTTP/1.1 200 OK\r\n"
"Server: " << BuildInfo::getFullVersionString() << "\r\n"
"Upgrade: Ripple/1.2\r\n"
"Connection: Upgrade\r\n"
"\r\n";
beast::http::write (write_buffer_, ss.str());
on_write_http_response(error_code(), 0);
}
else
{
std::stringstream ss;
ss <<
"HTTP/1.1 400 Bad Request\r\n"
"Server: " << BuildInfo::getFullVersionString() << "\r\n"
"\r\n"
"<html><head></head><body>"
"400 Bad Request<br>"
"The server requires an Upgrade request."
"</body></html>";
beast::http::write (write_buffer_, ss.str());
on_write_http_response(error_code(), 0);
}
return;
}
}
}
if (ec)
{
journal_.info <<
"on_read_http_request: " << ec.message();
detach("on_read_http_request");
return;
}
stream_.async_read_some (read_buffer_.prepare (Tuning::readBufferBytes),
strand_.wrap (std::bind (&PeerImp::on_read_http_request,
shared_from_this(), beast::asio::placeholders::error,
beast::asio::placeholders::bytes_transferred)));
}
beast::http::message
PeerImp::make_response (beast::http::message const& req)
{
beast::http::message resp;
// Unimplemented
return resp;
}
// Called repeatedly to send the bytes in the response
void
PeerImp::on_write_http_response (error_code ec, std::size_t bytes_transferred)
{
if (detaching_ || ec == boost::asio::error::operation_aborted)
return;
if (ec)
{
journal_.info <<
"on_write_http_response: " << ec.message();
detach("on_write_http_response");
return;
}
write_buffer_.consume (bytes_transferred);
if (write_buffer_.size() == 0)
{
do_protocol_start();
return;
}
stream_.async_write_some (write_buffer_.data(),
strand_.wrap (std::bind (&PeerImp::on_write_http_response,
shared_from_this(), beast::asio::placeholders::error,
beast::asio::placeholders::bytes_transferred)));
}
//------------------------------------------------------------------------------
// Protocol logic
// We have an encrypted connection to the peer.
// Have it say who it is so we know to avoid redundant connections.
// Establish that it really who we are talking to by having it sign a
// connection detail. Also need to establish no man in the middle attack
// is in progress.
void
PeerImp::do_protocol_start()
{
if (!sendHello ())
{
journal_.error << "Unable to send HELLO to " << remote_address_;
detach ("hello");
return;
}
on_read_protocol (error_code(), 0);
}
// Called repeatedly with protocol message data
void
PeerImp::on_read_protocol (error_code ec, std::size_t bytes_transferred)
{
if (detaching_ || ec == boost::asio::error::operation_aborted)
return;
if (! ec)
{
read_buffer_.commit (bytes_transferred);
ec = message_stream_.write (read_buffer_.data());
read_buffer_.consume (read_buffer_.size());
}
if (ec)
{
journal_.info <<
"on_read_protocol: " << ec.message();
detach("on_read_protocol");
return;
}
stream_.async_read_some (read_buffer_.prepare (Tuning::readBufferBytes),
strand_.wrap (std::bind (&PeerImp::on_read_protocol,
shared_from_this(), beast::asio::placeholders::error,
beast::asio::placeholders::bytes_transferred)));
}
// Called repeatedly to send protcol message data
void
PeerImp::on_write_protocol (error_code ec, std::size_t bytes_transferred)
{
// (this function isn't called yet)
if (detaching_ || ec == boost::asio::error::operation_aborted)
return;
}
void
PeerImp::handleShutdown (boost::system::error_code const& ec)
{
if (detaching_)
return;
if (ec == boost::asio::error::operation_aborted)
return;
if (ec)
{
journal_.info << "Shutdown: " << ec.message ();
detach ("hsd");
return;
}
}
void
PeerImp::handleWrite (boost::system::error_code const& ec, size_t bytes)
{
if (detaching_)
return;
// Call on IO strand
send_packet_.reset ();
if (ec == boost::asio::error::operation_aborted)
return;
if (detaching_)
return;
if (ec)
{
journal_.info << "Write: " << ec.message ();
detach ("hw");
return;
}
if (!send_queue_.empty ())
{
Message::pointer packet = send_queue_.front ();
if (packet)
{
sendForce (packet);
send_queue_.pop_front ();
}
}
}
void
PeerImp::handleVerifyTimer (boost::system::error_code const& ec)
{
if (detaching_)
return;
if (ec == boost::asio::error::operation_aborted)
{
// Timer canceled because deadline no longer needed.
}
else if (ec)
{
journal_.info << "Peer verify timer error";
}
else
{
// journal_.info << "Verify: Peer failed to verify in time.";
detach ("hvt");
}
}
//------------------------------------------------------------------------------
//
// abstract_protocol_handler
//
//------------------------------------------------------------------------------
PeerImp::error_code
PeerImp::on_message_unknown (std::uint16_t type)
{
error_code ec;
// TODO
return ec;
}
PeerImp::error_code
PeerImp::on_message_begin (std::uint16_t type,
std::shared_ptr <::google::protobuf::Message> const& m)
{
error_code ec;
#if 0
beast::debug_ostream log;
log << m->DebugString();
#endif
if (type == protocol::mtHELLO && state_ != stateConnected)
{
journal_.warning <<
"Unexpected TMHello";
ec = invalid_argument_error();
}
else if (type != protocol::mtHELLO && state_ == stateConnected)
{
journal_.warning <<
"Expected TMHello";
ec = invalid_argument_error();
}
if (! ec)
{
load_event_ = getApp().getJobQueue ().getLoadEventAP (
jtPEER, protocol_message_name(type));
}
return ec;
}
void
PeerImp::on_message_end (std::uint16_t,
std::shared_ptr <::google::protobuf::Message> const&)
{
load_event_.reset();
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMHello> const& m)
{
error_code ec;
timer_.cancel(ec);
std::uint32_t const ourTime (getApp().getOPs ().getNetworkTimeNC ());
std::uint32_t const minTime (ourTime - clockToleranceDeltaSeconds);
std::uint32_t const maxTime (ourTime + clockToleranceDeltaSeconds);
#ifdef BEAST_DEBUG
if (m->has_nettime ())
{
std::int64_t to = ourTime;
to -= m->nettime ();
journal_.debug <<
"Connect: time offset " << to;
}
#endif
auto protocol = BuildInfo::make_protocol(m->protoversion());
// VFALCO TODO Report these failures in the HTTP response
if (m->has_nettime () &&
((m->nettime () < minTime) || (m->nettime () > maxTime)))
{
if (m->nettime () > maxTime)
{
journal_.info <<
"Hello: Clock off by +" << m->nettime () - ourTime;
}
else if (m->nettime () < minTime)
{
journal_.info <<
"Hello: Clock off by -" << ourTime - m->nettime ();
}
}
else if (m->protoversionmin () > to_packed (BuildInfo::getCurrentProtocol()))
{
journal_.info <<
"Hello: Disconnect: Protocol mismatch [" <<
"Peer expects " << to_string (protocol) <<
" and we run " << to_string (BuildInfo::getCurrentProtocol()) << "]";
}
else if (! publicKey_.setNodePublic (m->nodepublic ()))
{
journal_.info <<
"Hello: Disconnect: Bad node public key.";
}
else if (! publicKey_.verifyNodePublic (
secureCookie_, m->nodeproof (), ECDSA::not_strict))
{
// Unable to verify they have private key for claimed public key.
journal_.info <<
"Hello: Disconnect: Failed to verify session.";
}
else
{
// Successful connection.
journal_.info <<
"Hello: Connect: " << publicKey_.humanNodePublic ();
if ((protocol != BuildInfo::getCurrentProtocol()) &&
journal_.active(beast::Journal::Severity::kInfo))
{
journal_.info <<
"Peer protocol: " << to_string (protocol);
}
hello_ = *m;
// Determine if this peer belongs to our cluster and get it's name
clusterNode_ = getApp().getUNL().nodeInCluster (
publicKey_, name_);
if (clusterNode_)
journal_.info <<
"Connected to cluster node " << name_;
assert (state_ == stateConnected);
// VFALCO TODO Remove this needless state
state_ = stateHandshaked;
auto const result = peerFinder_.activate (slot_,
RipplePublicKey (publicKey_), clusterNode_);
if (result == PeerFinder::Result::success)
{
state_ = stateActive;
assert(shortId_ == 0);
shortId_ = overlay_.next_id();
overlay_.activate(shared_from_this ());
// XXX Set timer: connection is in grace period to be useful.
// XXX Set timer: connection idle (idle may vary depending on connection type.)
if ((hello_.has_ledgerclosed ()) && (
hello_.ledgerclosed ().size () == (256 / 8)))
{
memcpy (closedLedgerHash_.begin (),
hello_.ledgerclosed ().data (), 256 / 8);
if ((hello_.has_ledgerprevious ()) &&
(hello_.ledgerprevious ().size () == (256 / 8)))
{
memcpy (previousLedgerHash_.begin (),
hello_.ledgerprevious ().data (), 256 / 8);
addLedger (previousLedgerHash_);
}
else
{
previousLedgerHash_.zero();
}
}
sendGetPeers();
return ec;
}
if (result == PeerFinder::Result::full)
{
// TODO Provide correct HTTP response
auto const redirects = peerFinder_.redirect (slot_);
send_endpoints (redirects.begin(), redirects.end());
}
else
{
// TODO Duplicate connection
}
}
// VFALCO Commented this out because we return an error code
// to the caller, who calls detach for us.
//publicKey_.clear ();
//detach ("recvh");
ec = invalid_argument_error();
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMPing> const& m)
{
error_code ec;
if (m->type () == protocol::TMPing::ptPING)
{
m->set_type (protocol::TMPing::ptPONG);
send (std::make_shared<Message> (*m, protocol::mtPING));
}
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMProofWork> const& m)
{
error_code ec;
if (m->has_response ())
{
// this is an answer to a proof of work we requested
if (m->response ().size () != (256 / 8))
{
charge (Resource::feeInvalidRequest);
return ec;
}
uint256 response;
memcpy (response.begin (), m->response ().data (), 256 / 8);
// VFALCO TODO Use a dependency injection here
PowResult r = getApp().getProofOfWorkFactory ().checkProof (
m->token (), response);
if (r == powOK)
{
// credit peer
// WRITEME
return ec;
}
// return error message
// WRITEME
if (r != powTOOEASY)
{
charge (Resource::feeBadProofOfWork);
}
return ec;
}
if (m->has_result ())
{
// this is a reply to a proof of work we sent
// WRITEME
}
if (m->has_target () && m->has_challenge () && m->has_iterations ())
{
// this is a challenge
// WRITEME: Reject from inbound connections
uint256 challenge, target;
if ((m->challenge ().size () != (256 / 8)) || (
m->target ().size () != (256 / 8)))
{
charge (Resource::feeInvalidRequest);
return ec;
}
memcpy (challenge.begin (), m->challenge ().data (), 256 / 8);
memcpy (target.begin (), m->target ().data (), 256 / 8);
ProofOfWork::pointer pow = std::make_shared<ProofOfWork> (
m->token (), m->iterations (), challenge, target);
if (!pow->isValid ())
{
charge (Resource::feeInvalidRequest);
return ec;
}
#if 0 // Until proof of work is completed, don't do it
getApp().getJobQueue ().addJob (
jtPROOFWORK,
"recvProof->doProof",
std::bind (&PeerImp::doProofOfWork, std::placeholders::_1,
std::weak_ptr <PeerImp> (shared_from_this ()), pow));
#endif
return ec;
}
p_journal_.info << "Bad proof of work";
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMCluster> const& m)
{
error_code ec;
if (!clusterNode_)
{
charge (Resource::feeUnwantedData);
return ec;
}
for (int i = 0; i < m->clusternodes().size(); ++i)
{
protocol::TMClusterNode const& node = m->clusternodes(i);
std::string name;
if (node.has_nodename())
name = node.nodename();
ClusterNodeStatus s(name, node.nodeload(), node.reporttime());
RippleAddress nodePub;
nodePub.setNodePublic(node.publickey());
getApp().getUNL().nodeUpdate(nodePub, s);
}
int loadSources = m->loadsources().size();
if (loadSources != 0)
{
Resource::Gossip gossip;
gossip.items.reserve (loadSources);
for (int i = 0; i < m->loadsources().size(); ++i)
{
protocol::TMLoadSource const& node = m->loadsources (i);
Resource::Gossip::Item item;
item.address = beast::IP::Endpoint::from_string (node.name());
item.balance = node.cost();
if (item.address != beast::IP::Endpoint())
gossip.items.push_back(item);
}
resourceManager_.importConsumers (name_, gossip);
}
getApp().getFeeTrack().setClusterFee(getApp().getUNL().getClusterFee());
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMGetPeers> const& m)
{
error_code ec;
// VFALCO TODO This message is now obsolete due to PeerFinder
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMPeers> const& m)
{
error_code ec;
// VFALCO TODO This message is now obsolete due to PeerFinder
std::vector <beast::IP::Endpoint> list;
list.reserve (m->nodes().size());
for (int i = 0; i < m->nodes ().size (); ++i)
{
in_addr addr;
addr.s_addr = m->nodes (i).ipv4 ();
{
beast::IP::AddressV4 v4 (ntohl (addr.s_addr));
beast::IP::Endpoint address (v4, m->nodes (i).ipv4port ());
list.push_back (address);
}
}
if (! list.empty())
peerFinder_.on_legacy_endpoints (list);
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMEndpoints> const& m)
{
error_code ec;
std::vector <PeerFinder::Endpoint> endpoints;
endpoints.reserve (m->endpoints().size());
for (int i = 0; i < m->endpoints ().size (); ++i)
{
PeerFinder::Endpoint endpoint;
protocol::TMEndpoint const& tm (m->endpoints(i));
// hops
endpoint.hops = tm.hops();
// ipv4
if (endpoint.hops > 0)
{
in_addr addr;
addr.s_addr = tm.ipv4().ipv4();
beast::IP::AddressV4 v4 (ntohl (addr.s_addr));
endpoint.address = beast::IP::Endpoint (v4, tm.ipv4().ipv4port ());
}
else
{
// This Endpoint describes the peer we are connected to.
// We will take the remote address seen on the socket and
// store that in the IP::Endpoint. If this is the first time,
// then we'll verify that their listener can receive incoming
// by performing a connectivity test.
//
endpoint.address = remote_address_.at_port (
tm.ipv4().ipv4port ());
}
endpoints.push_back (endpoint);
}
if (! endpoints.empty())
peerFinder_.on_endpoints (slot_, endpoints);
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMTransaction> const& m)
{
error_code ec;
Serializer s (m->rawtransaction ());
try
{
SerializerIterator sit (s);
SerializedTransaction::pointer stx = std::make_shared <
SerializedTransaction> (std::ref (sit));
uint256 txID = stx->getTransactionID ();
int flags;
if (! getApp().getHashRouter ().addSuppressionPeer (
txID, shortId_, flags))
{
// we have seen this transaction recently
if (flags & SF_BAD)
{
charge (Resource::feeInvalidSignature);
return ec;
}
if (!(flags & SF_RETRY))
return ec;
}
p_journal_.debug <<
"Got tx " << txID;
if (clusterNode_)
{
if (! m->has_deferred () || ! m->deferred ())
{
// Skip local checks if a server we trust
// put the transaction in its open ledger
flags |= SF_TRUSTED;
}
if (! getConfig().VALIDATION_PRIV.isSet())
{
// For now, be paranoid and have each validator
// check each transaction, regardless of source
flags |= SF_SIGGOOD;
}
}
if (getApp().getJobQueue().getJobCount(jtTRANSACTION) > 100)
p_journal_.info << "Transaction queue is full";
else if (getApp().getLedgerMaster().getValidatedLedgerAge() > 240)
p_journal_.trace << "No new transactions until synchronized";
else
getApp().getJobQueue ().addJob (jtTRANSACTION,
"recvTransaction->checkTransaction",
std::bind (
&PeerImp::checkTransaction, std::placeholders::_1,
flags, stx,
std::weak_ptr<PeerImp> (shared_from_this ())));
}
catch (...)
{
p_journal_.warning << "Transaction invalid: " <<
s.getHex();
}
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMGetLedger> const& m)
{
error_code ec;
getApp().getJobQueue().addJob (jtPACK, "recvGetLedger",
std::bind (&PeerImp::sGetLedger, std::weak_ptr<PeerImp> (
shared_from_this ()), m));
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMLedgerData> const& m)
{
error_code ec;
protocol::TMLedgerData& packet = *m;
if (m->nodes ().size () <= 0)
{
p_journal_.warning << "Ledger/TXset data with no nodes";
return ec;
}
if (m->has_requestcookie ())
{
Peer::ptr target = overlay_.findPeerByShortID (m->requestcookie ());
if (target)
{
m->clear_requestcookie ();
target->send (std::make_shared<Message> (
packet, protocol::mtLEDGER_DATA));
}
else
{
p_journal_.info << "Unable to route TX/ledger data reply";
charge (Resource::feeUnwantedData);
}
return ec;
}
uint256 hash;
if (m->ledgerhash ().size () != 32)
{
p_journal_.warning << "TX candidate reply with invalid hash size";
charge (Resource::feeInvalidRequest);
return ec;
}
memcpy (hash.begin (), m->ledgerhash ().data (), 32);
if (m->type () == protocol::liTS_CANDIDATE)
{
// got data for a candidate transaction set
getApp().getJobQueue().addJob (jtTXN_DATA, "recvPeerData",
std::bind (&PeerImp::peerTXData, std::placeholders::_1,
std::weak_ptr<PeerImp> (shared_from_this ()),
hash, m, p_journal_));
return ec;
}
if (!getApp().getInboundLedgers ().gotLedgerData (
hash, shared_from_this(), m))
{
p_journal_.trace << "Got data for unwanted ledger";
charge (Resource::feeUnwantedData);
}
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMProposeSet> const& m)
{
error_code ec;
protocol::TMProposeSet& set = *m;
// VFALCO Magic numbers are bad
if ((set.closetime() + 180) < getApp().getOPs().getCloseTimeNC())
return ec;
// VFALCO Magic numbers are bad
// Roll this into a validation function
if (
(set.currenttxhash ().size () != 32) ||
(set.nodepubkey ().size () < 28) ||
(set.signature ().size () < 56) ||
(set.nodepubkey ().size () > 128) ||
(set.signature ().size () > 128)
)
{
p_journal_.warning << "Proposal: malformed";
charge (Resource::feeInvalidSignature);
return ec;
}
if (set.has_previousledger () && (set.previousledger ().size () != 32))
{
p_journal_.warning << "Proposal: malformed";
charge (Resource::feeInvalidRequest);
return ec;
}
uint256 proposeHash, prevLedger;
memcpy (proposeHash.begin (), set.currenttxhash ().data (), 32);
if (set.has_previousledger ())
memcpy (prevLedger.begin (), set.previousledger ().data (), 32);
uint256 suppression = LedgerProposal::computeSuppressionID (
proposeHash, prevLedger, set.proposeseq(), set.closetime (),
Blob(set.nodepubkey ().begin (), set.nodepubkey ().end ()),
Blob(set.signature ().begin (), set.signature ().end ()));
if (! getApp().getHashRouter ().addSuppressionPeer (
suppression, shortId_))
{
p_journal_.trace << "Proposal: duplicate";
return ec;
}
RippleAddress signerPublic = RippleAddress::createNodePublic (
strCopy (set.nodepubkey ()));
if (signerPublic == getConfig ().VALIDATION_PUB)
{
p_journal_.trace << "Proposal: self";
return ec;
}
bool isTrusted = getApp().getUNL ().nodeInUNL (signerPublic);
if (!isTrusted && getApp().getFeeTrack ().isLoadedLocal ())
{
p_journal_.debug << "Proposal: Dropping UNTRUSTED (load)";
return ec;
}
p_journal_.trace <<
"Proposal: " << (isTrusted ? "trusted" : "UNTRUSTED");
uint256 consensusLCL;
{
Application::ScopedLockType lock (getApp ().getMasterLock ());
consensusLCL = getApp().getOPs ().getConsensusLCL ();
}
LedgerProposal::pointer proposal = std::make_shared<LedgerProposal> (
prevLedger.isNonZero () ? prevLedger : consensusLCL,
set.proposeseq (), proposeHash, set.closetime (),
signerPublic, suppression);
getApp().getJobQueue ().addJob (isTrusted ? jtPROPOSAL_t : jtPROPOSAL_ut,
"recvPropose->checkPropose", std::bind (
&PeerImp::checkPropose, std::placeholders::_1, &overlay_,
m, proposal, consensusLCL, publicKey_,
std::weak_ptr<PeerImp> (shared_from_this ()), clusterNode_,
p_journal_));
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMStatusChange> const& m)
{
error_code ec;
p_journal_.trace << "Status: Change";
if (!m->has_networktime ())
m->set_networktime (getApp().getOPs ().getNetworkTimeNC ());
if (!last_status_.has_newstatus () || m->has_newstatus ())
last_status_ = *m;
else
{
// preserve old status
protocol::NodeStatus status = last_status_.newstatus ();
last_status_ = *m;
m->set_newstatus (status);
}
if (m->newevent () == protocol::neLOST_SYNC)
{
if (!closedLedgerHash_.isZero ())
{
p_journal_.trace << "Status: Out of sync";
closedLedgerHash_.zero ();
}
previousLedgerHash_.zero ();
return ec;
}
if (m->has_ledgerhash () && (m->ledgerhash ().size () == (256 / 8)))
{
// a peer has changed ledgers
memcpy (closedLedgerHash_.begin (), m->ledgerhash ().data (), 256 / 8);
addLedger (closedLedgerHash_);
p_journal_.trace << "LCL is " << closedLedgerHash_;
}
else
{
p_journal_.trace << "Status: No ledger";
closedLedgerHash_.zero ();
}
if (m->has_ledgerhashprevious () &&
m->ledgerhashprevious ().size () == (256 / 8))
{
memcpy (previousLedgerHash_.begin (),
m->ledgerhashprevious ().data (), 256 / 8);
addLedger (previousLedgerHash_);
}
else previousLedgerHash_.zero ();
if (m->has_firstseq () && m->has_lastseq())
{
minLedger_ = m->firstseq ();
maxLedger_ = m->lastseq ();
// Work around some servers that report sequences incorrectly
if (minLedger_ == 0)
maxLedger_ = 0;
if (maxLedger_ == 0)
minLedger_ = 0;
}
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMHaveTransactionSet> const& m)
{
error_code ec;
uint256 hashes;
if (m->hash ().size () != (256 / 8))
{
charge (Resource::feeInvalidRequest);
return ec;
}
uint256 hash;
// VFALCO TODO There should be no use of memcpy() throughout the program.
// TODO Clean up this magic number
//
memcpy (hash.begin (), m->hash ().data (), 32);
if (m->status () == protocol::tsHAVE)
addTxSet (hash);
{
Application::ScopedLockType lock (getApp ().getMasterLock ());
if (!getApp().getOPs ().hasTXSet (
shared_from_this (), hash, m->status ()))
charge (Resource::feeUnwantedData);
}
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMValidation> const& m)
{
error_code ec;
std::uint32_t closeTime = getApp().getOPs().getCloseTimeNC();
if (m->validation ().size () < 50)
{
p_journal_.warning << "Validation: Too small";
charge (Resource::feeInvalidRequest);
return ec;
}
try
{
Serializer s (m->validation ());
SerializerIterator sit (s);
SerializedValidation::pointer val = std::make_shared <
SerializedValidation> (std::ref (sit), false);
if (closeTime > (120 + val->getFieldU32(sfSigningTime)))
{
p_journal_.trace << "Validation: Too old";
charge (Resource::feeUnwantedData);
return ec;
}
if (! getApp().getHashRouter ().addSuppressionPeer (
s.getSHA512Half(), shortId_))
{
p_journal_.trace << "Validation: duplicate";
return ec;
}
bool isTrusted = getApp().getUNL ().nodeInUNL (val->getSignerPublic ());
if (isTrusted || !getApp().getFeeTrack ().isLoadedLocal ())
{
getApp().getJobQueue ().addJob (
isTrusted ? jtVALIDATION_t : jtVALIDATION_ut,
"recvValidation->checkValidation",
std::bind (&PeerImp::checkValidation, std::placeholders::_1,
&overlay_, val, isTrusted, clusterNode_, m,
std::weak_ptr<PeerImp> (shared_from_this ()),
p_journal_));
}
else
{
p_journal_.debug <<
"Validation: Dropping UNTRUSTED (load)";
}
}
catch (std::exception const& e)
{
p_journal_.warning <<
"Validation: Exception, " << e.what();
charge (Resource::feeInvalidRequest);
}
catch (...)
{
p_journal_.warning <<
"Validation: Unknown exception";
charge (Resource::feeInvalidRequest);
}
return ec;
}
PeerImp::error_code
PeerImp::on_message (std::shared_ptr <protocol::TMGetObjectByHash> const& m)
{
error_code ec;
protocol::TMGetObjectByHash& packet = *m;
if (packet.query ())
{
// this is a query
if (packet.type () == protocol::TMGetObjectByHash::otFETCH_PACK)
{
doFetchPack (m);
return ec;
}
protocol::TMGetObjectByHash reply;
reply.set_query (false);
if (packet.has_seq ())
reply.set_seq (packet.seq ());
reply.set_type (packet.type ());
if (packet.has_ledgerhash ())
reply.set_ledgerhash (packet.ledgerhash ());
// This is a very minimal implementation
for (int i = 0; i < packet.objects_size (); ++i)
{
uint256 hash;
const protocol::TMIndexedObject& obj = packet.objects (i);
if (obj.has_hash () && (obj.hash ().size () == (256 / 8)))
{
memcpy (hash.begin (), obj.hash ().data (), 256 / 8);
// VFALCO TODO Move this someplace more sensible so we dont
// need to inject the NodeStore interfaces.
NodeObject::pointer hObj =
getApp().getNodeStore ().fetch (hash);
if (hObj)
{
protocol::TMIndexedObject& newObj = *reply.add_objects ();
newObj.set_hash (hash.begin (), hash.size ());
newObj.set_data (&hObj->getData ().front (),
hObj->getData ().size ());
if (obj.has_nodeid ())
newObj.set_index (obj.nodeid ());
if (!reply.has_seq () && (hObj->getLedgerIndex () != 0))
reply.set_seq (hObj->getLedgerIndex ());
}
}
}
p_journal_.trace <<
"GetObj: " << reply.objects_size () <<
" of " << packet.objects_size ();
send (std::make_shared<Message> (reply, protocol::mtGET_OBJECTS));
}
else
{
// this is a reply
std::uint32_t pLSeq = 0;
bool pLDo = true;
bool progress = false;
for (int i = 0; i < packet.objects_size (); ++i)
{
const protocol::TMIndexedObject& obj = packet.objects (i);
if (obj.has_hash () && (obj.hash ().size () == (256 / 8)))
{
if (obj.has_ledgerseq ())
{
if (obj.ledgerseq () != pLSeq)
{
if ((pLDo && (pLSeq != 0)) &&
p_journal_.active(beast::Journal::Severity::kDebug))
p_journal_.debug <<
"GetObj: Full fetch pack for " << pLSeq;
pLSeq = obj.ledgerseq ();
pLDo = !getApp().getOPs ().haveLedger (pLSeq);
if (!pLDo)
p_journal_.debug <<
"GetObj: Late fetch pack for " << pLSeq;
else
progress = true;
}
}
if (pLDo)
{
uint256 hash;
memcpy (hash.begin (), obj.hash ().data (), 256 / 8);
std::shared_ptr< Blob > data (
std::make_shared< Blob > (
obj.data ().begin (), obj.data ().end ()));
getApp().getOPs ().addFetchPack (hash, data);
}
}
}
if ((pLDo && (pLSeq != 0)) &&
p_journal_.active(beast::Journal::Severity::kDebug))
p_journal_.debug << "GetObj: Partial fetch pack for " << pLSeq;
if (packet.type () == protocol::TMGetObjectByHash::otFETCH_PACK)
getApp().getOPs ().gotFetchPack (progress, pLSeq);
}
return ec;
}
//------------------------------------------------------------------------------
// VFALCO NOTE This function is way too big and cumbersome.
void
PeerImp::getLedger (protocol::TMGetLedger& packet)
{
SHAMap::pointer map;
protocol::TMLedgerData reply;
bool fatLeaves = true, fatRoot = false;
if (packet.has_requestcookie ())
reply.set_requestcookie (packet.requestcookie ());
std::string logMe;
if (packet.itype () == protocol::liTS_CANDIDATE)
{
// Request is for a transaction candidate set
p_journal_.trace <<
"GetLedger: Tx candidate set";
if ((!packet.has_ledgerhash () || packet.ledgerhash ().size () != 32))
{
charge (Resource::feeInvalidRequest);
p_journal_.warning << "GetLedger: Tx candidate set invalid";
return;
}
uint256 txHash;
memcpy (txHash.begin (), packet.ledgerhash ().data (), 32);
{
Application::ScopedLockType lock (getApp ().getMasterLock ());
map = getApp().getOPs ().getTXMap (txHash);
}
if (!map)
{
if (packet.has_querytype () && !packet.has_requestcookie ())
{
p_journal_.debug <<
"GetLedger: Routing Tx set request";
struct get_usable_peers
{
typedef Overlay::PeerSequence return_type;
Overlay::PeerSequence usablePeers;
uint256 const& txHash;
Peer const* skip;
get_usable_peers(uint256 const& hash, Peer const* s)
: txHash(hash), skip(s)
{ }
void operator() (Peer::ptr const& peer)
{
if (peer->hasTxSet (txHash) && (peer.get () != skip))
usablePeers.push_back (peer);
}
return_type operator() ()
{
return usablePeers;
}
};
Overlay::PeerSequence usablePeers (overlay_.foreach (
get_usable_peers (txHash, this)));
if (usablePeers.empty ())
{
p_journal_.info <<
"GetLedger: Route TX set failed";
return;
}
Peer::ptr const& selectedPeer = usablePeers [
rand () % usablePeers.size ()];
packet.set_requestcookie (getShortId ());
selectedPeer->send (std::make_shared<Message> (
packet, protocol::mtGET_LEDGER));
return;
}
p_journal_.error <<
"GetLedger: Can't provide map ";
charge (Resource::feeInvalidRequest);
return;
}
reply.set_ledgerseq (0);
reply.set_ledgerhash (txHash.begin (), txHash.size ());
reply.set_type (protocol::liTS_CANDIDATE);
fatLeaves = false; // We'll already have most transactions
fatRoot = true; // Save a pass
}
else
{
if (getApp().getFeeTrack().isLoadedLocal() && !clusterNode_)
{
p_journal_.debug <<
"GetLedger: Too busy";
return;
}
// Figure out what ledger they want
p_journal_.trace <<
"GetLedger: Received";
Ledger::pointer ledger;
if (packet.has_ledgerhash ())
{
uint256 ledgerhash;
if (packet.ledgerhash ().size () != 32)
{
charge (Resource::feeInvalidRequest);
p_journal_.warning <<
"GetLedger: Invalid request";
return;
}
memcpy (ledgerhash.begin (), packet.ledgerhash ().data (), 32);
logMe += "LedgerHash:";
logMe += to_string (ledgerhash);
ledger = getApp().getLedgerMaster ().getLedgerByHash (ledgerhash);
if (!ledger && p_journal_.trace)
p_journal_.trace <<
"GetLedger: Don't have " << ledgerhash;
if (!ledger && (packet.has_querytype () &&
!packet.has_requestcookie ()))
{
std::uint32_t seq = 0;
if (packet.has_ledgerseq ())
seq = packet.ledgerseq ();
Overlay::PeerSequence peerList = overlay_.getActivePeers ();
Overlay::PeerSequence usablePeers;
for (auto const& peer : peerList)
{
if (peer->hasLedger (ledgerhash, seq) && (peer.get () != this))
usablePeers.push_back (peer);
}
if (usablePeers.empty ())
{
p_journal_.trace <<
"GetLedger: Cannot route";
return;
}
Peer::ptr const& selectedPeer = usablePeers [
rand () % usablePeers.size ()];
packet.set_requestcookie (getShortId ());
selectedPeer->send (
std::make_shared<Message> (packet, protocol::mtGET_LEDGER));
p_journal_.debug <<
"GetLedger: Request routed";
return;
}
}
else if (packet.has_ledgerseq ())
{
if (packet.ledgerseq() <
getApp().getLedgerMaster().getEarliestFetch())
{
p_journal_.debug <<
"GetLedger: Early ledger request";
return;
}
ledger = getApp().getLedgerMaster ().getLedgerBySeq (
packet.ledgerseq ());
if (!ledger && p_journal_.debug)
p_journal_.debug <<
"GetLedger: Don't have " << packet.ledgerseq ();
}
else if (packet.has_ltype () && (packet.ltype () == protocol::ltCURRENT))
{
ledger = getApp().getLedgerMaster ().getCurrentLedger ();
}
else if (packet.has_ltype () && (packet.ltype () == protocol::ltCLOSED) )
{
ledger = getApp().getLedgerMaster ().getClosedLedger ();
if (ledger && !ledger->isClosed ())
ledger = getApp().getLedgerMaster ().getLedgerBySeq (
ledger->getLedgerSeq () - 1);
}
else
{
charge (Resource::feeInvalidRequest);
p_journal_.warning <<
"GetLedger: Unknown request";
return;
}
if ((!ledger) || (packet.has_ledgerseq () && (
packet.ledgerseq () != ledger->getLedgerSeq ())))
{
charge (Resource::feeInvalidRequest);
if (p_journal_.warning && ledger)
p_journal_.warning <<
"GetLedger: Invalid sequence";
return;
}
if (!packet.has_ledgerseq() && (ledger->getLedgerSeq() <
getApp().getLedgerMaster().getEarliestFetch()))
{
p_journal_.debug <<
"GetLedger: Early ledger request";
return;
}
// Fill out the reply
uint256 lHash = ledger->getHash ();
reply.set_ledgerhash (lHash.begin (), lHash.size ());
reply.set_ledgerseq (ledger->getLedgerSeq ());
reply.set_type (packet.itype ());
if (packet.itype () == protocol::liBASE)
{
// they want the ledger base data
p_journal_.trace <<
"GetLedger: Base data";
Serializer nData (128);
ledger->addRaw (nData);
reply.add_nodes ()->set_nodedata (
nData.getDataPtr (), nData.getLength ());
SHAMap::pointer map = ledger->peekAccountStateMap ();
if (map && map->getHash ().isNonZero ())
{
// return account state root node if possible
Serializer rootNode (768);
if (map->getRootNode (rootNode, snfWIRE))
{
reply.add_nodes ()->set_nodedata (
rootNode.getDataPtr (), rootNode.getLength ());
if (ledger->getTransHash ().isNonZero ())
{
map = ledger->peekTransactionMap ();
if (map && map->getHash ().isNonZero ())
{
rootNode.erase ();
if (map->getRootNode (rootNode, snfWIRE))
reply.add_nodes ()->set_nodedata (
rootNode.getDataPtr (),
rootNode.getLength ());
}
}
}
}
Message::pointer oPacket = std::make_shared<Message> (
reply, protocol::mtLEDGER_DATA);
send (oPacket);
return;
}
if (packet.itype () == protocol::liTX_NODE)
{
map = ledger->peekTransactionMap ();
logMe += " TX:";
logMe += to_string (map->getHash ());
}
else if (packet.itype () == protocol::liAS_NODE)
{
map = ledger->peekAccountStateMap ();
logMe += " AS:";
logMe += to_string (map->getHash ());
}
}
if (!map || (packet.nodeids_size () == 0))
{
p_journal_.warning <<
"GetLedger: Can't find map or empty request";
charge (Resource::feeInvalidRequest);
return;
}
p_journal_.trace <<
"GetLeder: " << logMe;
for (int i = 0; i < packet.nodeids ().size (); ++i)
{
SHAMapNodeID mn (packet.nodeids (i).data (), packet.nodeids (i).size ());
if (!mn.isValid ())
{
p_journal_.warning <<
"GetLedger: Invalid node " << logMe;
charge (Resource::feeInvalidRequest);
return;
}
std::vector<SHAMapNodeID> nodeIDs;
std::list< Blob > rawNodes;
try
{
if (map->getNodeFat (mn, nodeIDs, rawNodes, fatRoot, fatLeaves))
{
assert (nodeIDs.size () == rawNodes.size ());
p_journal_.trace <<
"GetLedger: getNodeFat got " << rawNodes.size () << " nodes";
std::vector<SHAMapNodeID>::iterator nodeIDIterator;
std::list< Blob >::iterator rawNodeIterator;
for (nodeIDIterator = nodeIDs.begin (),
rawNodeIterator = rawNodes.begin ();
nodeIDIterator != nodeIDs.end ();
++nodeIDIterator, ++rawNodeIterator)
{
Serializer nID (33);
nodeIDIterator->addIDRaw (nID);
protocol::TMLedgerNode* node = reply.add_nodes ();
node->set_nodeid (nID.getDataPtr (), nID.getLength ());
node->set_nodedata (&rawNodeIterator->front (),
rawNodeIterator->size ());
}
}
else
p_journal_.warning <<
"GetLedger: getNodeFat returns false";
}
catch (std::exception&)
{
std::string info;
if (packet.itype () == protocol::liTS_CANDIDATE)
info = "TS candidate";
else if (packet.itype () == protocol::liBASE)
info = "Ledger base";
else if (packet.itype () == protocol::liTX_NODE)
info = "TX node";
else if (packet.itype () == protocol::liAS_NODE)
info = "AS node";
if (!packet.has_ledgerhash ())
info += ", no hash specified";
p_journal_.warning <<
"getNodeFat( " << mn << ") throws exception: " << info;
}
}
Message::pointer oPacket = std::make_shared<Message> (
reply, protocol::mtLEDGER_DATA);
send (oPacket);
}
//------------------------------------------------------------------------------
void
PeerImp::detach (const char* rsn)
{
if (! strand_.running_in_this_thread ())
{
strand_.post (std::bind (&PeerImp::detach,
shared_from_this (), rsn));
return;
}
if (!detaching_)
{
// NIKB TODO No - a race is NOT ok. This needs to be fixed
// to have PeerFinder work reliably.
detaching_ = true; // Race is ok.
peerFinder_.on_closed (slot_);
if (state_ == stateActive)
overlay_.onPeerDisconnect (shared_from_this ());
state_ = stateGracefulClose;
if (clusterNode_ && p_journal_.active(beast::Journal::Severity::kWarning))
journal_.warning <<
name_ << " left cluster: " << rsn;
send_queue_.clear ();
error_code ec;
timer_.cancel(ec);
stream_.next_layer().close(ec);
// VFALCO TODO Stop doing this.
if (publicKey_.isValid ())
publicKey_.clear (); // Be idempotent.
}
}
//--------------------------------------------------------------------------
void
PeerImp::sendGetPeers ()
{
// Ask peer for known other peers.
protocol::TMGetPeers msg;
msg.set_doweneedthis (1);
Message::pointer packet = std::make_shared<Message> (
msg, protocol::mtGET_PEERS);
send (packet);
}
void
PeerImp::charge (std::weak_ptr <PeerImp>& peer, Resource::Charge const& fee)
{
Peer::ptr p (peer.lock());
if (p != nullptr)
p->charge (fee);
}
void
PeerImp::sendForce (const Message::pointer& packet)
{
// must be on IO strand
if (!detaching_)
{
send_packet_ = packet;
boost::asio::async_write (stream_,
boost::asio::buffer (packet->getBuffer ()),
strand_.wrap (std::bind (
&PeerImp::handleWrite,
std::static_pointer_cast <PeerImp> (shared_from_this ()),
beast::asio::placeholders::error,
beast::asio::placeholders::bytes_transferred)));
}
}
bool
PeerImp::hashLatestFinishedMessage (const SSL *sslSession, unsigned char *hash,
size_t (*getFinishedMessage)(const SSL *, void *buf, size_t))
{
unsigned char buf[1024];
// Get our finished message and hash it.
std::memset(hash, 0, 64);
size_t len = getFinishedMessage (sslSession, buf, sizeof (buf));
if(len < sslMinimumFinishedLength)
return false;
SHA512 (buf, len, hash);
return true;
}
bool
PeerImp::calculateSessionCookie ()
{
SSL* ssl = stream_.native_handle();
if (!ssl)
{
journal_.error << "Cookie generation: No underlying connection";
return false;
}
unsigned char sha1[64];
unsigned char sha2[64];
if (!hashLatestFinishedMessage(ssl, sha1, SSL_get_finished))
{
journal_.error << "Cookie generation: local setup not complete";
return false;
}
if (!hashLatestFinishedMessage(ssl, sha2, SSL_get_peer_finished))
{
journal_.error << "Cookie generation: peer setup not complete";
return false;
}
// If both messages hash to the same value (i.e. match) something is
// wrong. This would cause the resulting cookie to be 0.
if (memcmp (sha1, sha2, sizeof (sha1)) == 0)
{
journal_.error << "Cookie generation: identical finished messages";
return false;
}
for (size_t i = 0; i < sizeof (sha1); ++i)
sha1[i] ^= sha2[i];
// Finally, derive the actual cookie for the values that we have
// calculated.
secureCookie_ = Serializer::getSHA512Half (sha1, sizeof(sha1));
return true;
}
bool
PeerImp::sendHello ()
{
if (!calculateSessionCookie())
return false;
Blob vchSig;
getApp().getLocalCredentials ().getNodePrivate ().signNodePrivate (
secureCookie_, vchSig);
protocol::TMHello h;
h.set_protoversion (to_packed (BuildInfo::getCurrentProtocol()));
h.set_protoversionmin (to_packed (BuildInfo::getMinimumProtocol()));
h.set_fullversion (BuildInfo::getFullVersionString ());
h.set_nettime (getApp().getOPs ().getNetworkTimeNC ());
h.set_nodepublic (getApp().getLocalCredentials ().getNodePublic (
).humanNodePublic ());
h.set_nodeproof (&vchSig[0], vchSig.size ());
// DEPRECATED
h.set_ipv4port (overlay_.serverHandler().setup().overlay.port);
h.set_testnet (false);
// We always advertise ourselves as private in the HELLO message. This
// suppresses the old peer advertising code and allows PeerFinder to
// take over the functionality.
h.set_nodeprivate (true);
Ledger::pointer closedLedger = getApp().getLedgerMaster ().getClosedLedger ();
if (closedLedger && closedLedger->isClosed ())
{
uint256 hash = closedLedger->getHash ();
h.set_ledgerclosed (hash.begin (), hash.size ());
hash = closedLedger->getParentHash ();
h.set_ledgerprevious (hash.begin (), hash.size ());
}
Message::pointer packet = std::make_shared<Message> (
h, protocol::mtHELLO);
send (packet);
return true;
}
void
PeerImp::addLedger (uint256 const& hash)
{
std::lock_guard<std::mutex> sl(recentLock_);
if (std::find (recentLedgers_.begin(),
recentLedgers_.end(), hash) != recentLedgers_.end())
return;
// VFALCO TODO See if a sorted vector would be better.
if (recentLedgers_.size () == 128)
recentLedgers_.pop_front ();
recentLedgers_.push_back (hash);
}
void
PeerImp::addTxSet (uint256 const& hash)
{
std::lock_guard<std::mutex> sl(recentLock_);
if (std::find (recentTxSets_.begin (),
recentTxSets_.end (), hash) != recentTxSets_.end ())
return;
if (recentTxSets_.size () == 128)
recentTxSets_.pop_front ();
recentTxSets_.push_back (hash);
}
void
PeerImp::doFetchPack (const std::shared_ptr<protocol::TMGetObjectByHash>& packet)
{
// VFALCO TODO Invert this dependency using an observer and shared state object.
// Don't queue fetch pack jobs if we're under load or we already have
// some queued.
if (getApp().getFeeTrack ().isLoadedLocal () ||
(getApp().getLedgerMaster().getValidatedLedgerAge() > 40) ||
(getApp().getJobQueue().getJobCount(jtPACK) > 10))
{
p_journal_.info << "Too busy to make fetch pack";
return;
}
if (packet->ledgerhash ().size () != 32)
{
p_journal_.warning << "FetchPack hash size malformed";
charge (Resource::feeInvalidRequest);
return;
}
uint256 hash;
memcpy (hash.begin (), packet->ledgerhash ().data (), 32);
getApp().getJobQueue ().addJob (jtPACK, "MakeFetchPack",
std::bind (&NetworkOPs::makeFetchPack, &getApp().getOPs (),
std::placeholders::_1, std::weak_ptr<PeerImp> (shared_from_this ()),
packet, hash, UptimeTimer::getInstance ().getElapsedSeconds ()));
}
void
PeerImp::doProofOfWork (Job&, std::weak_ptr <PeerImp> peer,
ProofOfWork::pointer pow)
{
if (peer.expired ())
return;
uint256 solution = pow->solve ();
if (solution.isZero ())
{
p_journal_.warning << "Failed to solve proof of work";
}
else
{
Peer::ptr pptr (peer.lock ());
if (pptr)
{
protocol::TMProofWork reply;
reply.set_token (pow->getToken ());
reply.set_response (solution.begin (), solution.size ());
pptr->send (std::make_shared<Message> (
reply, protocol::mtPROOFOFWORK));
}
else
{
// WRITEME: Save solved proof of work for new connection
}
}
}
void
PeerImp::checkTransaction (Job&, int flags,
SerializedTransaction::pointer stx, std::weak_ptr<PeerImp> peer)
{
// VFALCO TODO Rewrite to not use exceptions
try
{
// Expired?
if (stx->isFieldPresent(sfLastLedgerSequence) &&
(stx->getFieldU32 (sfLastLedgerSequence) <
getApp().getLedgerMaster().getValidLedgerIndex()))
{
getApp().getHashRouter().setFlag(stx->getTransactionID(), SF_BAD);
charge (peer, Resource::feeUnwantedData);
return;
}
auto validate = (flags & SF_SIGGOOD) ? Validate::NO : Validate::YES;
auto tx = std::make_shared<Transaction> (stx, validate);
if (tx->getStatus () == INVALID)
{
getApp().getHashRouter ().setFlag (stx->getTransactionID (), SF_BAD);
charge (peer, Resource::feeInvalidSignature);
return;
}
else
getApp().getHashRouter ().setFlag (stx->getTransactionID (), SF_SIGGOOD);
bool const trusted (flags & SF_TRUSTED);
getApp().getOPs ().processTransaction (tx, trusted, false, false);
}
catch (...)
{
getApp().getHashRouter ().setFlag (stx->getTransactionID (), SF_BAD);
charge (peer, Resource::feeInvalidRequest);
}
}
// Called from our JobQueue
void
PeerImp::checkPropose (Job& job, Overlay* pPeers,
std::shared_ptr <protocol::TMProposeSet> packet,
LedgerProposal::pointer proposal, uint256 consensusLCL,
RippleAddress nodePublic, std::weak_ptr<PeerImp> peer,
bool fromCluster, beast::Journal journal)
{
bool sigGood = false;
bool isTrusted = (job.getType () == jtPROPOSAL_t);
journal.trace <<
"Checking " << (isTrusted ? "trusted" : "UNTRUSTED") << " proposal";
assert (packet);
protocol::TMProposeSet& set = *packet;
uint256 prevLedger;
if (set.has_previousledger ())
{
// proposal includes a previous ledger
journal.trace <<
"proposal with previous ledger";
memcpy (prevLedger.begin (), set.previousledger ().data (), 256 / 8);
if (!fromCluster && !proposal->checkSign (set.signature ()))
{
Peer::ptr p = peer.lock ();
journal.warning <<
"Proposal with previous ledger fails sig check";
charge (peer, Resource::feeInvalidSignature);
return;
}
else
sigGood = true;
}
else
{
if (consensusLCL.isNonZero () && proposal->checkSign (set.signature ()))
{
prevLedger = consensusLCL;
sigGood = true;
}
else
{
// Could be mismatched prev ledger
journal.warning <<
"Ledger proposal fails signature check";
proposal->setSignature (set.signature ());
}
}
if (isTrusted)
{
getApp().getOPs ().processTrustedProposal (
proposal, packet, nodePublic, prevLedger, sigGood);
}
else if (sigGood && (prevLedger == consensusLCL))
{
// relay untrusted proposal
journal.trace <<
"relaying UNTRUSTED proposal";
std::set<Peer::ShortId> peers;
if (getApp().getHashRouter ().swapSet (
proposal->getSuppressionID (), peers, SF_RELAYED))
{
pPeers->foreach (send_if_not (
std::make_shared<Message> (set, protocol::mtPROPOSE_LEDGER),
peer_in_set(peers)));
}
}
else
{
journal.debug <<
"Not relaying UNTRUSTED proposal";
}
}
void
PeerImp::checkValidation (Job&, Overlay* pPeers,
SerializedValidation::pointer val, bool isTrusted, bool isCluster,
std::shared_ptr<protocol::TMValidation> packet,
std::weak_ptr<PeerImp> peer, beast::Journal journal)
{
try
{
uint256 signingHash = val->getSigningHash();
if (!isCluster && !val->isValid (signingHash))
{
journal.warning <<
"Validation is invalid";
charge (peer, Resource::feeInvalidRequest);
return;
}
std::string source;
PeerImp::ptr lp = peer.lock ();
if (lp)
source = std::to_string(lp->getShortId());
else
source = "unknown";
std::set<Peer::ShortId> peers;
//----------------------------------------------------------------------
//
{
SerializedValidation const& sv (*val);
Validators::ReceivedValidation rv;
rv.ledgerHash = sv.getLedgerHash ();
rv.publicKey = sv.getSignerPublic();
getApp ().getValidators ().on_receive_validation (rv);
}
//
//----------------------------------------------------------------------
if (getApp().getOPs ().recvValidation (val, source) &&
getApp().getHashRouter ().swapSet (
signingHash, peers, SF_RELAYED))
{
pPeers->foreach (send_if_not (
std::make_shared<Message> (*packet, protocol::mtVALIDATION),
peer_in_set(peers)));
}
}
catch (...)
{
journal.trace <<
"Exception processing validation";
charge (peer, Resource::feeInvalidRequest);
}
}
// This is dispatched by the job queue
void
PeerImp::sGetLedger (std::weak_ptr<PeerImp> wPeer,
std::shared_ptr <protocol::TMGetLedger> packet)
{
std::shared_ptr<PeerImp> peer = wPeer.lock ();
if (peer)
peer->getLedger (*packet);
}
// VFALCO TODO Make this non-static
void
PeerImp::peerTXData (Job&, std::weak_ptr <PeerImp> wPeer, uint256 const& hash,
std::shared_ptr <protocol::TMLedgerData> pPacket, beast::Journal journal)
{
std::shared_ptr <PeerImp> peer = wPeer.lock ();
if (!peer)
return;
protocol::TMLedgerData& packet = *pPacket;
std::list<SHAMapNodeID> nodeIDs;
std::list< Blob > nodeData;
for (int i = 0; i < packet.nodes ().size (); ++i)
{
const protocol::TMLedgerNode& node = packet.nodes (i);
if (!node.has_nodeid () || !node.has_nodedata () || (
node.nodeid ().size () != 33))
{
journal.warning << "LedgerData request with invalid node ID";
peer->charge (Resource::feeInvalidRequest);
return;
}
nodeIDs.push_back (SHAMapNodeID {node.nodeid ().data (),
static_cast<int>(node.nodeid ().size ())});
nodeData.push_back (Blob (node.nodedata ().begin (),
node.nodedata ().end ()));
}
SHAMapAddNode san;
{
Application::ScopedLockType lock (getApp ().getMasterLock ());
san = getApp().getOPs().gotTXData (peer, hash, nodeIDs, nodeData);
}
if (san.isInvalid ())
{
peer->charge (Resource::feeUnwantedData);
}
}
} // ripple
Reference in New Issue View Git Blame Copy Permalink