Enforce a 20s timeout when making validator list requests (RIPD-1737)

This commit is contained in:
Mike Ellery
2019-04-08 12:09:04 -07:00
committed by Nik Bougalis
parent 2e26377e7c
commit dd99bf479f
4 changed files with 329 additions and 146 deletions

View File

@@ -58,7 +58,8 @@ namespace ripple {
@li @c "version": 1
@li @c "refreshInterval" (optional)
@li @c "refreshInterval" (optional, integer minutes).
This value is clamped internally to [1,1440] (1 min - 1 day)
*/
class ValidatorSite
{
@@ -125,11 +126,15 @@ private:
// The configured list of URIs for fetching lists
std::vector<Site> sites_;
// time to allow for requests to complete
const std::chrono::seconds requestTimeout_;
public:
ValidatorSite (
boost::asio::io_service& ios,
ValidatorList& validators,
beast::Journal j);
beast::Journal j,
std::chrono::seconds timeout = std::chrono::seconds{20});
~ValidatorSite ();
/** Load configured site URIs.
@@ -184,8 +189,15 @@ public:
private:
/// Queue next site to be fetched
/// lock over state_mutex_ required
void
setTimer ();
setTimer (std::lock_guard<std::mutex>&);
/// request took too long
void
onRequestTimeout (
std::size_t siteIdx,
error_code const& ec);
/// Fetch site whose time has come
void

View File

@@ -26,15 +26,15 @@
#include <ripple/basics/Slice.h>
#include <ripple/json/json_reader.h>
#include <ripple/protocol/JsonFields.h>
#include <boost/algorithm/clamp.hpp>
#include <boost/regex.hpp>
#include <algorithm>
namespace ripple {
// default site query frequency - 5 minutes
auto constexpr DEFAULT_REFRESH_INTERVAL = std::chrono::minutes{5};
auto constexpr ERROR_RETRY_INTERVAL = std::chrono::seconds{30};
unsigned short constexpr MAX_REDIRECTS = 3;
auto constexpr default_refresh_interval = std::chrono::minutes{5};
auto constexpr error_retry_interval = std::chrono::seconds{30};
unsigned short constexpr max_redirects = 3;
ValidatorSite::Site::Resource::Resource (std::string uri_)
: uri {std::move(uri_)}
@@ -82,7 +82,7 @@ ValidatorSite::Site::Site (std::string uri)
: loadedResource {std::make_shared<Resource>(std::move(uri))}
, startingResource {loadedResource}
, redirCount {0}
, refreshInterval {DEFAULT_REFRESH_INTERVAL}
, refreshInterval {default_refresh_interval}
, nextRefresh {clock_type::now()}
{
}
@@ -90,7 +90,8 @@ ValidatorSite::Site::Site (std::string uri)
ValidatorSite::ValidatorSite (
boost::asio::io_service& ios,
ValidatorList& validators,
beast::Journal j)
beast::Journal j,
std::chrono::seconds timeout)
: ios_ (ios)
, validators_ (validators)
, j_ (j)
@@ -98,6 +99,7 @@ ValidatorSite::ValidatorSite (
, fetching_ (false)
, pending_ (false)
, stopping_ (false)
, requestTimeout_ (timeout)
{
}
@@ -153,7 +155,7 @@ ValidatorSite::start ()
{
std::lock_guard <std::mutex> lock{state_mutex_};
if (timer_.expires_at() == clock_type::time_point{})
setTimer ();
setTimer (lock);
}
void
@@ -168,20 +170,29 @@ ValidatorSite::stop()
{
std::unique_lock<std::mutex> lock{state_mutex_};
stopping_ = true;
cv_.wait(lock, [&]{ return ! fetching_; });
// work::cancel() must be called before the
// cv wait in order to kick any asio async operations
// that might be pending.
if(auto sp = work_.lock())
sp->cancel();
cv_.wait(lock, [&]{ return ! fetching_; });
error_code ec;
timer_.cancel(ec);
// docs indicate cancel() can throw, but this should be
// reconsidered if it changes to noexcept
try
{
timer_.cancel();
}
catch (boost::system::system_error const&)
{
}
stopping_ = false;
pending_ = false;
cv_.notify_all();
}
void
ValidatorSite::setTimer ()
ValidatorSite::setTimer (std::lock_guard<std::mutex>& state_lock)
{
std::lock_guard <std::mutex> lock{sites_mutex_};
@@ -196,8 +207,11 @@ ValidatorSite::setTimer ()
pending_ = next->nextRefresh <= clock_type::now();
cv_.notify_all();
timer_.expires_at (next->nextRefresh);
timer_.async_wait (std::bind (&ValidatorSite::onTimer, this,
std::distance (sites_.begin (), next), std::placeholders::_1));
auto idx = std::distance (sites_.begin (), next);
timer_.async_wait ([this, idx] (boost::system::error_code const& ec)
{
this->onTimer (idx, ec);
});
}
}
@@ -205,22 +219,42 @@ void
ValidatorSite::makeRequest (
std::shared_ptr<Site::Resource> resource,
std::size_t siteIdx,
std::lock_guard<std::mutex>& lock)
std::lock_guard<std::mutex>& sites_lock)
{
fetching_ = true;
sites_[siteIdx].activeResource = resource;
std::shared_ptr<detail::Work> sp;
auto onFetch =
[this, siteIdx] (error_code const& err, detail::response_type&& resp)
auto timeoutCancel =
[this] ()
{
std::lock_guard <std::mutex> lock_state{state_mutex_};
// docs indicate cancel_one() can throw, but this
// should be reconsidered if it changes to noexcept
try
{
timer_.cancel_one();
}
catch (boost::system::system_error const&)
{
}
};
auto onFetch =
[this, siteIdx, timeoutCancel] (
error_code const& err, detail::response_type&& resp)
{
timeoutCancel ();
onSiteFetch (err, std::move(resp), siteIdx);
};
auto onFetchFile =
[this, siteIdx] (error_code const& err, std::string const& resp)
{
onTextFetch (err, resp, siteIdx);
};
[this, siteIdx, timeoutCancel] (
error_code const& err, std::string const& resp)
{
timeoutCancel ();
onTextFetch (err, resp, siteIdx);
};
JLOG (j_.debug()) << "Starting request for " << resource->uri;
if (resource->pUrl.scheme == "https")
{
@@ -252,6 +286,34 @@ ValidatorSite::makeRequest (
work_ = sp;
sp->run ();
// start a timer for the request, which shouldn't take more
// than requestTimeout_ to complete
std::lock_guard <std::mutex> lock_state{state_mutex_};
timer_.expires_after (requestTimeout_);
timer_.async_wait ([this, siteIdx] (boost::system::error_code const& ec)
{
this->onRequestTimeout (siteIdx, ec);
});
}
void
ValidatorSite::onRequestTimeout (
std::size_t siteIdx,
error_code const& ec)
{
if (ec)
return;
{
std::lock_guard <std::mutex> lock_site{sites_mutex_};
JLOG (j_.warn()) <<
"Request for " << sites_[siteIdx].activeResource->uri <<
" took too long";
}
std::lock_guard<std::mutex> lock_state{state_mutex_};
if(auto sp = work_.lock())
sp->cancel();
}
void
@@ -268,14 +330,12 @@ ValidatorSite::onTimer (
return;
}
std::lock_guard <std::mutex> lock{sites_mutex_};
sites_[siteIdx].nextRefresh =
clock_type::now() + sites_[siteIdx].refreshInterval;
assert(! fetching_);
sites_[siteIdx].redirCount = 0;
try
{
std::lock_guard <std::mutex> lock{sites_mutex_};
sites_[siteIdx].nextRefresh =
clock_type::now() + sites_[siteIdx].refreshInterval;
sites_[siteIdx].redirCount = 0;
// the WorkSSL client can throw if SSL init fails
makeRequest(sites_[siteIdx].startingResource, siteIdx, lock);
}
@@ -292,7 +352,7 @@ void
ValidatorSite::parseJsonResponse (
std::string const& res,
std::size_t siteIdx,
std::lock_guard<std::mutex>& lock)
std::lock_guard<std::mutex>& sites_lock)
{
Json::Reader r;
Json::Value body;
@@ -370,10 +430,15 @@ ValidatorSite::parseJsonResponse (
if (body.isMember ("refresh_interval") &&
body["refresh_interval"].isNumeric ())
{
// TODO: should we sanity check/clamp this value
// to something reasonable?
sites_[siteIdx].refreshInterval =
std::chrono::minutes{body["refresh_interval"].asUInt ()};
using namespace std::chrono_literals;
std::chrono::minutes const refresh =
boost::algorithm::clamp(
std::chrono::minutes {body["refresh_interval"].asUInt ()},
1min,
24h);
sites_[siteIdx].refreshInterval = refresh;
sites_[siteIdx].nextRefresh =
clock_type::now() + sites_[siteIdx].refreshInterval;
}
}
@@ -381,7 +446,7 @@ std::shared_ptr<ValidatorSite::Site::Resource>
ValidatorSite::processRedirect (
detail::response_type& res,
std::size_t siteIdx,
std::lock_guard<std::mutex>& lock)
std::lock_guard<std::mutex>& sites_lock)
{
using namespace boost::beast::http;
std::shared_ptr<Site::Resource> newLocation;
@@ -395,7 +460,7 @@ ValidatorSite::processRedirect (
throw std::runtime_error{"missing location"};
}
if (sites_[siteIdx].redirCount == MAX_REDIRECTS)
if (sites_[siteIdx].redirCount == max_redirects)
{
JLOG (j_.warn()) <<
"Exceeded max redirects for validator list at " <<
@@ -435,6 +500,8 @@ ValidatorSite::onSiteFetch(
{
{
std::lock_guard <std::mutex> lock_sites{sites_mutex_};
JLOG (j_.debug()) << "Got completion for "
<< sites_[siteIdx].activeResource->uri;
auto onError = [&](std::string const& errMsg, bool retry)
{
sites_[siteIdx].lastRefreshStatus.emplace(
@@ -443,7 +510,7 @@ ValidatorSite::onSiteFetch(
errMsg});
if (retry)
sites_[siteIdx].nextRefresh =
clock_type::now() + ERROR_RETRY_INTERVAL;
clock_type::now() + error_retry_interval;
};
if (ec)
{
@@ -506,7 +573,7 @@ ValidatorSite::onSiteFetch(
std::lock_guard <std::mutex> lock_state{state_mutex_};
fetching_ = false;
if (! stopping_)
setTimer ();
setTimer (lock_state);
cv_.notify_all();
}
@@ -547,7 +614,7 @@ ValidatorSite::onTextFetch(
std::lock_guard <std::mutex> lock_state{state_mutex_};
fetching_ = false;
if (! stopping_)
setTimer ();
setTimer (lock_state);
cv_.notify_all();
}

View File

@@ -30,8 +30,10 @@
#include <test/jtx.h>
#include <test/jtx/TrustedPublisherServer.h>
#include <test/unit_test/FileDirGuard.h>
#include <boost/algorithm/string/join.hpp>
#include <boost/algorithm/string/predicate.hpp>
#include <boost/asio.hpp>
#include <boost/range/adaptor/transformed.hpp>
#include <chrono>
namespace ripple {
@@ -48,6 +50,8 @@ constexpr const char* realValidatorContents()
}
)vl";
}
auto constexpr default_expires = std::chrono::seconds{3600};
}
class ValidatorSite_test : public beast::unit_test::suite
@@ -185,13 +189,24 @@ private:
}
};
void
testFetchList (
std::vector<std::pair<std::string, std::string>> const& paths)
struct FetchListConfig
{
testcase << "Fetch list - " << paths[0].first <<
(paths.size() > 1 ? ", " + paths[1].first : "");
std::string path;
std::string msg;
bool failFetch = false;
bool failApply = false;
int serverVersion = 1;
std::chrono::seconds expiresFromNow = detail::default_expires;
int expectedRefreshMin = 0;
};
void
testFetchList (std::vector<FetchListConfig> const& paths)
{
testcase << "Fetch list - " <<
boost::algorithm::join (paths |
boost::adaptors::transformed(
[](FetchListConfig const& cfg){ return cfg.path; }),
", ");
using namespace jtx;
Env env (*this);
@@ -204,20 +219,16 @@ private:
std::vector<std::string> emptyCfgKeys;
struct publisher
{
publisher(FetchListConfig const& c) : cfg{c} {}
std::unique_ptr<TrustedPublisherServer> server;
std::vector<Validator> list;
std::string uri;
std::string expectMsg;
bool shouldFail;
FetchListConfig const& cfg;
bool isRetry;
};
std::vector<publisher> servers;
auto const sequence = 1;
auto const version = 1;
using namespace std::chrono_literals;
NetClock::time_point const expiration =
env.timeKeeper().now() + 3600s;
auto constexpr listSize = 20;
std::vector<std::string> cfgPublishers;
@@ -233,11 +244,9 @@ private:
publisherPublic, publisherSecret,
pubSigningKeys.first, pubSigningKeys.second, 1);
servers.push_back({});
servers.push_back(cfg);
auto& item = servers.back();
item.shouldFail = ! cfg.second.empty();
item.isRetry = cfg.first == "/bad-resource";
item.expectMsg = cfg.second;
item.isRetry = cfg.path == "/bad-resource";
item.list.reserve (listSize);
while (item.list.size () < listSize)
item.list.push_back (randomValidator());
@@ -247,20 +256,24 @@ private:
pubSigningKeys,
manifest,
sequence,
expiration,
version,
env.timeKeeper().now() + cfg.expiresFromNow,
cfg.serverVersion,
item.list);
std::stringstream uri;
uri << "http://" << item.server->local_endpoint() << cfg.first;
uri << "http://" << item.server->local_endpoint() << cfg.path;
item.uri = uri.str();
}
BEAST_EXPECT(trustedKeys.load (
emptyLocalKey, emptyCfgKeys, cfgPublishers));
using namespace std::chrono_literals;
auto sites = std::make_unique<ValidatorSite> (
env.app().getIOService(), env.app().validators(), journal);
env.app().getIOService(),
env.app().validators(),
journal,
2s);
std::vector<std::string> uris;
for (auto const& u : servers)
@@ -269,30 +282,45 @@ private:
sites->start();
sites->join();
auto const jv = sites->getJson();
for (auto const& u : servers)
{
for (auto const& val : u.list)
{
BEAST_EXPECT(
trustedKeys.listed (val.masterPublic) != u.shouldFail);
trustedKeys.listed (val.masterPublic) != u.cfg.failApply);
BEAST_EXPECT(
trustedKeys.listed (val.signingPublic) != u.shouldFail);
trustedKeys.listed (val.signingPublic) != u.cfg.failApply);
}
auto const jv = sites->getJson();
Json::Value myStatus;
for (auto const& vs : jv[jss::validator_sites])
if (vs[jss::uri].asString().find(u.uri) != std::string::npos)
myStatus = vs;
BEAST_EXPECTS(
myStatus[jss::last_refresh_message].asString().empty()
!= u.shouldFail, to_string(myStatus));
if (u.shouldFail)
!= u.cfg.failFetch,
to_string(myStatus) + "\n" + sink.strm_.str());
if (! u.cfg.msg.empty())
{
BEAST_EXPECTS(
sink.strm_.str().find(u.cfg.msg) != std::string::npos,
sink.strm_.str());
}
if (u.cfg.expectedRefreshMin)
{
BEAST_EXPECTS(
myStatus[jss::refresh_interval_min].asInt()
== u.cfg.expectedRefreshMin,
to_string(myStatus));
}
if (u.cfg.failFetch)
{
using namespace std::chrono;
BEAST_EXPECTS(
sink.strm_.str().find(u.expectMsg) != std::string::npos,
sink.strm_.str());
log << " -- Msg: " <<
myStatus[jss::last_refresh_message].asString() << std::endl;
std::stringstream nextRefreshStr
@@ -412,40 +440,99 @@ public:
testConfigLoad ();
// fetch single site
testFetchList ({{"/validators",""}});
testFetchList ({{"/validators", ""}});
// fetch multiple sites
testFetchList ({{"/validators",""}, {"/validators",""}});
testFetchList ({{"/validators", ""}, {"/validators", ""}});
// fetch single site with single redirects
testFetchList ({{"/redirect_once/301",""}});
testFetchList ({{"/redirect_once/302",""}});
testFetchList ({{"/redirect_once/307",""}});
testFetchList ({{"/redirect_once/308",""}});
testFetchList ({{"/redirect_once/301", ""}});
testFetchList ({{"/redirect_once/302", ""}});
testFetchList ({{"/redirect_once/307", ""}});
testFetchList ({{"/redirect_once/308", ""}});
// one redirect, one not
testFetchList ({{"/validators",""}, {"/redirect_once/302",""}});
testFetchList ({{"/validators", ""}, {"/redirect_once/302", ""}});
// fetch single site with undending redirect (fails to load)
testFetchList ({{"/redirect_forever/301", "Exceeded max redirects"}});
testFetchList ({
{"/redirect_forever/301", "Exceeded max redirects", true, true}});
// two that redirect forever
testFetchList ({
{"/redirect_forever/307","Exceeded max redirects"},
{"/redirect_forever/308","Exceeded max redirects"}});
{"/redirect_forever/307", "Exceeded max redirects", true, true},
{"/redirect_forever/308", "Exceeded max redirects", true, true}});
// one undending redirect, one not
testFetchList (
{{"/validators",""},
{"/redirect_forever/302","Exceeded max redirects"}});
{{"/validators", ""},
{"/redirect_forever/302", "Exceeded max redirects", true, true}});
// invalid redir Location
testFetchList ({
{"/redirect_to/ftp://invalid-url/302",
"Invalid redirect location"}});
"Invalid redirect location",
true,
true}});
testFetchList ({
{"/redirect_to/file://invalid-url/302",
"Invalid redirect location",
true,
true}});
// invalid json
testFetchList ({{"/validators/bad", "Unable to parse JSON response"}});
testFetchList ({
{"/validators/bad", "Unable to parse JSON response", true, true}});
// error status returned
testFetchList ({{"/bad-resource", "returned bad status"}});
testFetchList ({
{"/bad-resource", "returned bad status", true, true}});
// location field missing
testFetchList ({
{"/redirect_nolo/308", "returned a redirect with no Location"}});
{"/redirect_nolo/308",
"returned a redirect with no Location",
true,
true}});
// json fields missing
testFetchList ({
{"/validators/missing", "Missing fields in JSON response"}});
{"/validators/missing",
"Missing fields in JSON response",
true,
true}});
// timeout
testFetchList ({
{"/sleep/3", "took too long", true, true}});
// bad manifest version
testFetchList ({
{"/validators", "Unsupported version", false, true, 4}});
using namespace std::chrono_literals;
// get old validator list
testFetchList ({
{"/validators", "Stale validator list", false, true, 1, 0s}});
// force an out-of-range expiration value
testFetchList ({
{"/validators",
"Invalid validator list",
false,
true,
1,
std::chrono::seconds{Json::Value::maxInt + 1}}});
// verify refresh intervals are properly clamped
testFetchList ({
{"/validators/refresh/0",
"",
false,
false,
1,
detail::default_expires,
1}}); // minimum of 1 minute
testFetchList ({
{"/validators/refresh/10",
"",
false,
false,
1,
detail::default_expires,
10}}); // 10 minutes is fine
testFetchList ({
{"/validators/refresh/2000",
"",
false,
false,
1,
detail::default_expires,
60*24}}); // max of 24 hours
testFileURLs();
}
};

View File

@@ -28,6 +28,8 @@
#include <boost/asio.hpp>
#include <boost/algorithm/string/predicate.hpp>
#include <boost/beast/http.hpp>
#include <boost/lexical_cast.hpp>
#include <thread>
namespace ripple {
namespace test {
@@ -44,8 +46,7 @@ class TrustedPublisherServer
socket_type sock_;
boost::asio::ip::tcp::acceptor acceptor_;
std::string list_;
std::function<std::string(int)> getList_;
public:
@@ -82,14 +83,16 @@ public:
data.pop_back();
data += "]}";
std::string blob = base64_encode(data);
list_ = "{\"blob\":\"" + blob + "\"";
auto const sig = sign(keys.first, keys.second, makeSlice(data));
list_ += ",\"signature\":\"" + strHex(sig) + "\"";
list_ += ",\"manifest\":\"" + manifest + "\"";
list_ += ",\"version\":" + std::to_string(version) + '}';
getList_ = [blob, sig, manifest, version](int interval) {
std::stringstream l;
l << "{\"blob\":\"" << blob << "\"" <<
",\"signature\":\"" << strHex(sig) << "\"" <<
",\"manifest\":\"" << manifest << "\"" <<
",\"refresh_interval\": " << interval <<
",\"version\":" << version << '}';
return l.str();
};
acceptor_.open(ep.protocol());
error_code ec;
@@ -163,61 +166,75 @@ private:
error_code ec;
for (;;)
{
req_type req;
http::read(sock, sb, req, ec);
if (ec)
break;
auto path = req.target().to_string();
resp_type res;
res.insert("Server", "TrustedPublisherServer");
res.version(req.version());
if (boost::starts_with(path, "/validators"))
{
res.result(http::status::ok);
res.insert("Content-Type", "application/json");
if (path == "/validators/bad")
res.body() = "{ 'bad': \"1']" ;
else if (path == "/validators/missing")
res.body() = "{\"version\": 1}";
else
res.body() = list_;
}
else if (boost::starts_with(path, "/redirect"))
{
if (boost::ends_with(path, "/301"))
res.result(http::status::moved_permanently);
else if (boost::ends_with(path, "/302"))
res.result(http::status::found);
else if (boost::ends_with(path, "/307"))
res.result(http::status::temporary_redirect);
else if (boost::ends_with(path, "/308"))
res.result(http::status::permanent_redirect);
std::stringstream location;
if (boost::starts_with(path, "/redirect_to/"))
{
location << path.substr(13);
}
else if (! boost::starts_with(path, "/redirect_nolo"))
{
location << "http://" << local_endpoint() <<
(boost::starts_with(path, "/redirect_forever/") ?
path : "/validators");
}
if (! location.str().empty())
res.insert("Location", location.str());
}
else
{
// unknown request
res.result(boost::beast::http::status::not_found);
res.insert("Content-Type", "text/html");
res.body() = "The file '" + path + "' was not found";
}
req_type req;
try
{
http::read(sock, sb, req, ec);
if (ec)
break;
auto path = req.target().to_string();
res.insert("Server", "TrustedPublisherServer");
res.version(req.version());
if (boost::starts_with(path, "/validators"))
{
res.result(http::status::ok);
res.insert("Content-Type", "application/json");
if (path == "/validators/bad")
res.body() = "{ 'bad': \"1']" ;
else if (path == "/validators/missing")
res.body() = "{\"version\": 1}";
else
{
int refresh = 5;
if (boost::starts_with(path, "/validators/refresh"))
refresh =
boost::lexical_cast<unsigned int>(
path.substr(20));
res.body() = getList_(refresh);
}
}
else if (boost::starts_with(path, "/sleep/"))
{
auto const sleep_sec =
boost::lexical_cast<unsigned int>(path.substr(7));
std::this_thread::sleep_for(
std::chrono::seconds{sleep_sec});
}
else if (boost::starts_with(path, "/redirect"))
{
if (boost::ends_with(path, "/301"))
res.result(http::status::moved_permanently);
else if (boost::ends_with(path, "/302"))
res.result(http::status::found);
else if (boost::ends_with(path, "/307"))
res.result(http::status::temporary_redirect);
else if (boost::ends_with(path, "/308"))
res.result(http::status::permanent_redirect);
std::stringstream location;
if (boost::starts_with(path, "/redirect_to/"))
{
location << path.substr(13);
}
else if (! boost::starts_with(path, "/redirect_nolo"))
{
location << "http://" << local_endpoint() <<
(boost::starts_with(path, "/redirect_forever/") ?
path : "/validators");
}
if (! location.str().empty())
res.insert("Location", location.str());
}
else
{
// unknown request
res.result(boost::beast::http::status::not_found);
res.insert("Content-Type", "text/html");
res.body() = "The file '" + path + "' was not found";
}
res.prepare_payload();
}
catch (std::exception const& e)