diff options
Diffstat (limited to 'src/rpcserver.cpp')
| -rw-r--r-- | src/rpcserver.cpp | 554 |
1 files changed, 41 insertions, 513 deletions
diff --git a/src/rpcserver.cpp b/src/rpcserver.cpp index 9362401b1e..5d7e2125e5 100644 --- a/src/rpcserver.cpp +++ b/src/rpcserver.cpp @@ -12,13 +12,9 @@ #include "ui_interface.h" #include "util.h" #include "utilstrencodings.h" -#ifdef ENABLE_WALLET -#include "wallet/wallet.h" -#endif -#include <boost/algorithm/string.hpp> -#include <boost/asio.hpp> -#include <boost/asio/ssl.hpp> +#include "univalue/univalue.h" + #include <boost/bind.hpp> #include <boost/filesystem.hpp> #include <boost/foreach.hpp> @@ -27,28 +23,20 @@ #include <boost/shared_ptr.hpp> #include <boost/signals2/signal.hpp> #include <boost/thread.hpp> +#include <boost/algorithm/string/case_conv.hpp> // for to_upper() -#include "univalue/univalue.h" - -using namespace boost::asio; using namespace RPCServer; using namespace std; -static std::string strRPCUserColonPass; - static bool fRPCRunning = false; static bool fRPCInWarmup = true; static std::string rpcWarmupStatus("RPC server started"); static CCriticalSection cs_rpcWarmup; - -//! These are created by StartRPCThreads, destroyed in StopRPCThreads -static boost::asio::io_service* rpc_io_service = NULL; -static map<string, boost::shared_ptr<deadline_timer> > deadlineTimers; -static ssl::context* rpc_ssl_context = NULL; -static boost::thread_group* rpc_worker_group = NULL; -static boost::asio::io_service::work *rpc_dummy_work = NULL; -static std::vector<CSubNet> rpc_allow_subnets; //!< List of subnets to allow RPC connections from -static std::vector< boost::shared_ptr<ip::tcp::acceptor> > rpc_acceptors; +/* Timer-creating functions */ +static std::vector<RPCTimerInterface*> timerInterfaces; +/* Map of name to timer. + * @note Can be changed to std::unique_ptr when C++11 */ +static std::map<std::string, boost::shared_ptr<RPCTimerBase> > deadlineTimers; static struct CRPCSignals { @@ -169,7 +157,6 @@ vector<unsigned char> ParseHexO(const UniValue& o, string strKey) return ParseHexV(find_value(o, strKey), strKey); } - /** * Note: This interface may still be subject to change. */ @@ -261,8 +248,6 @@ UniValue stop(const UniValue& params, bool fHelp) return "Bitcoin server stopping"; } - - /** * Call Table */ @@ -399,7 +384,7 @@ CRPCTable::CRPCTable() } } -const CRPCCommand *CRPCTable::operator[](const std::string& name) const +const CRPCCommand *CRPCTable::operator[](const std::string &name) const { map<string, const CRPCCommand*>::const_iterator it = mapCommands.find(name); if (it == mapCommands.end()) @@ -407,373 +392,26 @@ const CRPCCommand *CRPCTable::operator[](const std::string& name) const return (*it).second; } - -bool HTTPAuthorized(map<string, string>& mapHeaders) -{ - string strAuth = mapHeaders["authorization"]; - if (strAuth.substr(0,6) != "Basic ") - return false; - string strUserPass64 = strAuth.substr(6); boost::trim(strUserPass64); - string strUserPass = DecodeBase64(strUserPass64); - return TimingResistantEqual(strUserPass, strRPCUserColonPass); -} - -void ErrorReply(std::ostream& stream, const UniValue& objError, const UniValue& id) -{ - // Send error reply from json-rpc error object - int nStatus = HTTP_INTERNAL_SERVER_ERROR; - int code = find_value(objError, "code").get_int(); - if (code == RPC_INVALID_REQUEST) nStatus = HTTP_BAD_REQUEST; - else if (code == RPC_METHOD_NOT_FOUND) nStatus = HTTP_NOT_FOUND; - string strReply = JSONRPCReply(NullUniValue, objError, id); - stream << HTTPReply(nStatus, strReply, false) << std::flush; -} - -CNetAddr BoostAsioToCNetAddr(boost::asio::ip::address address) -{ - CNetAddr netaddr; - // Make sure that IPv4-compatible and IPv4-mapped IPv6 addresses are treated as IPv4 addresses - if (address.is_v6() - && (address.to_v6().is_v4_compatible() - || address.to_v6().is_v4_mapped())) - address = address.to_v6().to_v4(); - - if(address.is_v4()) - { - boost::asio::ip::address_v4::bytes_type bytes = address.to_v4().to_bytes(); - netaddr.SetRaw(NET_IPV4, &bytes[0]); - } - else - { - boost::asio::ip::address_v6::bytes_type bytes = address.to_v6().to_bytes(); - netaddr.SetRaw(NET_IPV6, &bytes[0]); - } - return netaddr; -} - -bool ClientAllowed(const boost::asio::ip::address& address) -{ - CNetAddr netaddr = BoostAsioToCNetAddr(address); - BOOST_FOREACH(const CSubNet &subnet, rpc_allow_subnets) - if (subnet.Match(netaddr)) - return true; - return false; -} - -template <typename Protocol> -class AcceptedConnectionImpl : public AcceptedConnection -{ -public: - AcceptedConnectionImpl( - boost::asio::io_service& io_service, - ssl::context &context, - bool fUseSSL) : - sslStream(io_service, context), - _d(sslStream, fUseSSL), - _stream(_d) - { - } - - virtual std::iostream& stream() - { - return _stream; - } - - virtual std::string peer_address_to_string() const - { - return peer.address().to_string(); - } - - virtual void close() - { - _stream.close(); - } - - typename Protocol::endpoint peer; - boost::asio::ssl::stream<typename Protocol::socket> sslStream; - -private: - SSLIOStreamDevice<Protocol> _d; - boost::iostreams::stream< SSLIOStreamDevice<Protocol> > _stream; -}; - -void ServiceConnection(AcceptedConnection *conn); - -//! Forward declaration required for RPCListen -template <typename Protocol, typename SocketAcceptorService> -static void RPCAcceptHandler(boost::shared_ptr< basic_socket_acceptor<Protocol, SocketAcceptorService> > acceptor, - ssl::context& context, - bool fUseSSL, - boost::shared_ptr< AcceptedConnection > conn, - const boost::system::error_code& error); - -/** - * Sets up I/O resources to accept and handle a new connection. - */ -template <typename Protocol, typename SocketAcceptorService> -static void RPCListen(boost::shared_ptr< basic_socket_acceptor<Protocol, SocketAcceptorService> > acceptor, - ssl::context& context, - const bool fUseSSL) -{ - // Accept connection - boost::shared_ptr< AcceptedConnectionImpl<Protocol> > conn(new AcceptedConnectionImpl<Protocol>(acceptor->get_io_service(), context, fUseSSL)); - - acceptor->async_accept( - conn->sslStream.lowest_layer(), - conn->peer, - boost::bind(&RPCAcceptHandler<Protocol, SocketAcceptorService>, - acceptor, - boost::ref(context), - fUseSSL, - conn, - _1)); -} - - -/** - * Accept and handle incoming connection. - */ -template <typename Protocol, typename SocketAcceptorService> -static void RPCAcceptHandler(boost::shared_ptr< basic_socket_acceptor<Protocol, SocketAcceptorService> > acceptor, - ssl::context& context, - const bool fUseSSL, - boost::shared_ptr< AcceptedConnection > conn, - const boost::system::error_code& error) +bool StartRPC() { - // Immediately start accepting new connections, except when we're cancelled or our socket is closed. - if (error != boost::asio::error::operation_aborted && acceptor->is_open()) - RPCListen(acceptor, context, fUseSSL); - - AcceptedConnectionImpl<ip::tcp>* tcp_conn = dynamic_cast< AcceptedConnectionImpl<ip::tcp>* >(conn.get()); - - if (error) - { - // TODO: Actually handle errors - LogPrintf("%s: Error: %s\n", __func__, error.message()); - } - // Restrict callers by IP. It is important to - // do this before starting client thread, to filter out - // certain DoS and misbehaving clients. - else if (tcp_conn && !ClientAllowed(tcp_conn->peer.address())) - { - // Only send a 403 if we're not using SSL to prevent a DoS during the SSL handshake. - if (!fUseSSL) - conn->stream() << HTTPError(HTTP_FORBIDDEN, false) << std::flush; - conn->close(); - } - else { - ServiceConnection(conn.get()); - conn->close(); - } -} - -static ip::tcp::endpoint ParseEndpoint(const std::string &strEndpoint, int defaultPort) -{ - std::string addr; - int port = defaultPort; - SplitHostPort(strEndpoint, port, addr); - return ip::tcp::endpoint(boost::asio::ip::address::from_string(addr), port); -} - -void StartRPCThreads() -{ - rpc_allow_subnets.clear(); - rpc_allow_subnets.push_back(CSubNet("127.0.0.0/8")); // always allow IPv4 local subnet - rpc_allow_subnets.push_back(CSubNet("::1")); // always allow IPv6 localhost - if (mapMultiArgs.count("-rpcallowip")) - { - const vector<string>& vAllow = mapMultiArgs["-rpcallowip"]; - BOOST_FOREACH(string strAllow, vAllow) - { - CSubNet subnet(strAllow); - if(!subnet.IsValid()) - { - uiInterface.ThreadSafeMessageBox( - strprintf("Invalid -rpcallowip subnet specification: %s. Valid are a single IP (e.g. 1.2.3.4), a network/netmask (e.g. 1.2.3.4/255.255.255.0) or a network/CIDR (e.g. 1.2.3.4/24).", strAllow), - "", CClientUIInterface::MSG_ERROR); - StartShutdown(); - return; - } - rpc_allow_subnets.push_back(subnet); - } - } - std::string strAllowed; - BOOST_FOREACH(const CSubNet &subnet, rpc_allow_subnets) - strAllowed += subnet.ToString() + " "; - LogPrint("rpc", "Allowing RPC connections from: %s\n", strAllowed); - - if (mapArgs["-rpcpassword"] == "") - { - LogPrintf("No rpcpassword set - using random cookie authentication\n"); - if (!GenerateAuthCookie(&strRPCUserColonPass)) { - uiInterface.ThreadSafeMessageBox( - _("Error: A fatal internal error occurred, see debug.log for details"), // Same message as AbortNode - "", CClientUIInterface::MSG_ERROR); - StartShutdown(); - return; - } - } else { - strRPCUserColonPass = mapArgs["-rpcuser"] + ":" + mapArgs["-rpcpassword"]; - } - - assert(rpc_io_service == NULL); - rpc_io_service = new boost::asio::io_service(); - rpc_ssl_context = new ssl::context(*rpc_io_service, ssl::context::sslv23); - - const bool fUseSSL = GetBoolArg("-rpcssl", false); - - if (fUseSSL) - { - rpc_ssl_context->set_options(ssl::context::no_sslv2 | ssl::context::no_sslv3); - - boost::filesystem::path pathCertFile(GetArg("-rpcsslcertificatechainfile", "server.cert")); - if (!pathCertFile.is_complete()) pathCertFile = boost::filesystem::path(GetDataDir()) / pathCertFile; - if (boost::filesystem::exists(pathCertFile)) rpc_ssl_context->use_certificate_chain_file(pathCertFile.string()); - else LogPrintf("ThreadRPCServer ERROR: missing server certificate file %s\n", pathCertFile.string()); - - boost::filesystem::path pathPKFile(GetArg("-rpcsslprivatekeyfile", "server.pem")); - if (!pathPKFile.is_complete()) pathPKFile = boost::filesystem::path(GetDataDir()) / pathPKFile; - if (boost::filesystem::exists(pathPKFile)) rpc_ssl_context->use_private_key_file(pathPKFile.string(), ssl::context::pem); - else LogPrintf("ThreadRPCServer ERROR: missing server private key file %s\n", pathPKFile.string()); - - string strCiphers = GetArg("-rpcsslciphers", "TLSv1.2+HIGH:TLSv1+HIGH:!SSLv2:!aNULL:!eNULL:!3DES:@STRENGTH"); - SSL_CTX_set_cipher_list(rpc_ssl_context->impl(), strCiphers.c_str()); - } - - std::vector<ip::tcp::endpoint> vEndpoints; - bool bBindAny = false; - int defaultPort = GetArg("-rpcport", BaseParams().RPCPort()); - if (!mapArgs.count("-rpcallowip")) // Default to loopback if not allowing external IPs - { - vEndpoints.push_back(ip::tcp::endpoint(boost::asio::ip::address_v6::loopback(), defaultPort)); - vEndpoints.push_back(ip::tcp::endpoint(boost::asio::ip::address_v4::loopback(), defaultPort)); - if (mapArgs.count("-rpcbind")) - { - LogPrintf("WARNING: option -rpcbind was ignored because -rpcallowip was not specified, refusing to allow everyone to connect\n"); - } - } else if (mapArgs.count("-rpcbind")) // Specific bind address - { - BOOST_FOREACH(const std::string &addr, mapMultiArgs["-rpcbind"]) - { - try { - vEndpoints.push_back(ParseEndpoint(addr, defaultPort)); - } - catch (const boost::system::system_error&) - { - uiInterface.ThreadSafeMessageBox( - strprintf(_("Could not parse -rpcbind value %s as network address"), addr), - "", CClientUIInterface::MSG_ERROR); - StartShutdown(); - return; - } - } - } else { // No specific bind address specified, bind to any - vEndpoints.push_back(ip::tcp::endpoint(boost::asio::ip::address_v6::any(), defaultPort)); - vEndpoints.push_back(ip::tcp::endpoint(boost::asio::ip::address_v4::any(), defaultPort)); - // Prefer making the socket dual IPv6/IPv4 instead of binding - // to both addresses separately. - bBindAny = true; - } - - bool fListening = false; - std::string strerr; - std::string straddress; - BOOST_FOREACH(const ip::tcp::endpoint &endpoint, vEndpoints) - { - try { - boost::asio::ip::address bindAddress = endpoint.address(); - straddress = bindAddress.to_string(); - LogPrintf("Binding RPC on address %s port %i (IPv4+IPv6 bind any: %i)\n", straddress, endpoint.port(), bBindAny); - boost::system::error_code v6_only_error; - boost::shared_ptr<ip::tcp::acceptor> acceptor(new ip::tcp::acceptor(*rpc_io_service)); - - acceptor->open(endpoint.protocol()); - acceptor->set_option(boost::asio::ip::tcp::acceptor::reuse_address(true)); - - // Try making the socket dual IPv6/IPv4 when listening on the IPv6 "any" address - acceptor->set_option(boost::asio::ip::v6_only( - !bBindAny || bindAddress != boost::asio::ip::address_v6::any()), v6_only_error); - - acceptor->bind(endpoint); - acceptor->listen(socket_base::max_connections); - - RPCListen(acceptor, *rpc_ssl_context, fUseSSL); - - fListening = true; - rpc_acceptors.push_back(acceptor); - // If dual IPv6/IPv4 bind successful, skip binding to IPv4 separately - if(bBindAny && bindAddress == boost::asio::ip::address_v6::any() && !v6_only_error) - break; - } - catch (const boost::system::system_error& e) - { - LogPrintf("ERROR: Binding RPC on address %s port %i failed: %s\n", straddress, endpoint.port(), e.what()); - strerr = strprintf(_("An error occurred while setting up the RPC address %s port %u for listening: %s"), straddress, endpoint.port(), e.what()); - } - } - - if (!fListening) { - uiInterface.ThreadSafeMessageBox(strerr, "", CClientUIInterface::MSG_ERROR); - StartShutdown(); - return; - } - - rpc_worker_group = new boost::thread_group(); - for (int i = 0; i < GetArg("-rpcthreads", 4); i++) - rpc_worker_group->create_thread(boost::bind(&boost::asio::io_service::run, rpc_io_service)); + LogPrint("rpc", "Starting RPC\n"); fRPCRunning = true; g_rpcSignals.Started(); + return true; } -void StartDummyRPCThread() +void InterruptRPC() { - if(rpc_io_service == NULL) - { - rpc_io_service = new boost::asio::io_service(); - /* Create dummy "work" to keep the thread from exiting when no timeouts active, - * see http://www.boost.org/doc/libs/1_51_0/doc/html/boost_asio/reference/io_service.html#boost_asio.reference.io_service.stopping_the_io_service_from_running_out_of_work */ - rpc_dummy_work = new boost::asio::io_service::work(*rpc_io_service); - rpc_worker_group = new boost::thread_group(); - rpc_worker_group->create_thread(boost::bind(&boost::asio::io_service::run, rpc_io_service)); - fRPCRunning = true; - } + LogPrint("rpc", "Interrupting RPC\n"); + // Interrupt e.g. running longpolls + fRPCRunning = false; } -void StopRPCThreads() +void StopRPC() { - if (rpc_io_service == NULL) return; - // Set this to false first, so that longpolling loops will exit when woken up - fRPCRunning = false; - - // First, cancel all timers and acceptors - // This is not done automatically by ->stop(), and in some cases the destructor of - // boost::asio::io_service can hang if this is skipped. - boost::system::error_code ec; - BOOST_FOREACH(const boost::shared_ptr<ip::tcp::acceptor> &acceptor, rpc_acceptors) - { - acceptor->cancel(ec); - if (ec) - LogPrintf("%s: Warning: %s when cancelling acceptor\n", __func__, ec.message()); - } - rpc_acceptors.clear(); - BOOST_FOREACH(const PAIRTYPE(std::string, boost::shared_ptr<deadline_timer>) &timer, deadlineTimers) - { - timer.second->cancel(ec); - if (ec) - LogPrintf("%s: Warning: %s when cancelling timer\n", __func__, ec.message()); - } + LogPrint("rpc", "Stopping RPC\n"); deadlineTimers.clear(); - - DeleteAuthCookie(); - - rpc_io_service->stop(); g_rpcSignals.Stopped(); - if (rpc_worker_group != NULL) - rpc_worker_group->join_all(); - delete rpc_dummy_work; rpc_dummy_work = NULL; - delete rpc_worker_group; rpc_worker_group = NULL; - delete rpc_ssl_context; rpc_ssl_context = NULL; - delete rpc_io_service; rpc_io_service = NULL; } bool IsRPCRunning() @@ -802,36 +440,6 @@ bool RPCIsInWarmup(std::string *outStatus) return fRPCInWarmup; } -void RPCRunHandler(const boost::system::error_code& err, boost::function<void(void)> func) -{ - if (!err) - func(); -} - -void RPCRunLater(const std::string& name, boost::function<void(void)> func, int64_t nSeconds) -{ - assert(rpc_io_service != NULL); - - if (deadlineTimers.count(name) == 0) - { - deadlineTimers.insert(make_pair(name, - boost::shared_ptr<deadline_timer>(new deadline_timer(*rpc_io_service)))); - } - deadlineTimers[name]->expires_from_now(boost::posix_time::seconds(nSeconds)); - deadlineTimers[name]->async_wait(boost::bind(RPCRunHandler, _1, func)); -} - -class JSONRequest -{ -public: - UniValue id; - string strMethod; - UniValue params; - - JSONRequest() { id = NullUniValue; } - void parse(const UniValue& valRequest); -}; - void JSONRequest::parse(const UniValue& valRequest) { // Parse request @@ -862,7 +470,6 @@ void JSONRequest::parse(const UniValue& valRequest) throw JSONRPCError(RPC_INVALID_REQUEST, "Params must be an array"); } - static UniValue JSONRPCExecOne(const UniValue& req) { UniValue rpc_result(UniValue::VOBJ); @@ -887,7 +494,7 @@ static UniValue JSONRPCExecOne(const UniValue& req) return rpc_result; } -static string JSONRPCExecBatch(const UniValue& vReq) +std::string JSONRPCExecBatch(const UniValue& vReq) { UniValue ret(UniValue::VARR); for (unsigned int reqIdx = 0; reqIdx < vReq.size(); reqIdx++) @@ -896,107 +503,6 @@ static string JSONRPCExecBatch(const UniValue& vReq) return ret.write() + "\n"; } -static bool HTTPReq_JSONRPC(AcceptedConnection *conn, - string& strRequest, - map<string, string>& mapHeaders, - bool fRun) -{ - // Check authorization - if (mapHeaders.count("authorization") == 0) - { - conn->stream() << HTTPError(HTTP_UNAUTHORIZED, false) << std::flush; - return false; - } - - if (!HTTPAuthorized(mapHeaders)) - { - LogPrintf("ThreadRPCServer incorrect password attempt from %s\n", conn->peer_address_to_string()); - /* Deter brute-forcing - We don't support exposing the RPC port, so this shouldn't result - in a DoS. */ - MilliSleep(250); - - conn->stream() << HTTPError(HTTP_UNAUTHORIZED, false) << std::flush; - return false; - } - - JSONRequest jreq; - try - { - // Parse request - UniValue valRequest; - if (!valRequest.read(strRequest)) - throw JSONRPCError(RPC_PARSE_ERROR, "Parse error"); - - string strReply; - - // singleton request - if (valRequest.isObject()) { - jreq.parse(valRequest); - - UniValue result = tableRPC.execute(jreq.strMethod, jreq.params); - - // Send reply - strReply = JSONRPCReply(result, NullUniValue, jreq.id); - - // array of requests - } else if (valRequest.isArray()) - strReply = JSONRPCExecBatch(valRequest.get_array()); - else - throw JSONRPCError(RPC_PARSE_ERROR, "Top-level object parse error"); - - conn->stream() << HTTPReplyHeader(HTTP_OK, fRun, strReply.size()) << strReply << std::flush; - } - catch (const UniValue& objError) - { - ErrorReply(conn->stream(), objError, jreq.id); - return false; - } - catch (const std::exception& e) - { - ErrorReply(conn->stream(), JSONRPCError(RPC_PARSE_ERROR, e.what()), jreq.id); - return false; - } - return true; -} - -void ServiceConnection(AcceptedConnection *conn) -{ - bool fRun = true; - while (fRun && !ShutdownRequested()) - { - int nProto = 0; - map<string, string> mapHeaders; - string strRequest, strMethod, strURI; - - // Read HTTP request line - if (!ReadHTTPRequestLine(conn->stream(), nProto, strMethod, strURI)) - break; - - // Read HTTP message headers and body - ReadHTTPMessage(conn->stream(), mapHeaders, strRequest, nProto, MAX_SIZE); - - // HTTP Keep-Alive is false; close connection immediately - if ((mapHeaders["connection"] == "close") || (!GetBoolArg("-rpckeepalive", true))) - fRun = false; - - // Process via JSON-RPC API - if (strURI == "/") { - if (!HTTPReq_JSONRPC(conn, strRequest, mapHeaders, fRun)) - break; - - // Process via HTTP REST API - } else if (strURI.substr(0, 6) == "/rest/" && GetBoolArg("-rest", false)) { - if (!HTTPReq_REST(conn, strURI, strRequest, mapHeaders, fRun)) - break; - - } else { - conn->stream() << HTTPError(HTTP_NOT_FOUND, false) << std::flush; - break; - } - } -} - UniValue CRPCTable::execute(const std::string &strMethod, const UniValue ¶ms) const { // Return immediately if in warmup @@ -1037,4 +543,26 @@ std::string HelpExampleRpc(const std::string& methodname, const std::string& arg "\"method\": \"" + methodname + "\", \"params\": [" + args + "] }' -H 'content-type: text/plain;' http://127.0.0.1:8332/\n"; } +void RPCRegisterTimerInterface(RPCTimerInterface *iface) +{ + timerInterfaces.push_back(iface); +} + +void RPCUnregisterTimerInterface(RPCTimerInterface *iface) +{ + std::vector<RPCTimerInterface*>::iterator i = std::find(timerInterfaces.begin(), timerInterfaces.end(), iface); + assert(i != timerInterfaces.end()); + timerInterfaces.erase(i); +} + +void RPCRunLater(const std::string& name, boost::function<void(void)> func, int64_t nSeconds) +{ + if (timerInterfaces.empty()) + throw JSONRPCError(RPC_INTERNAL_ERROR, "No timer handler registered for RPC"); + deadlineTimers.erase(name); + RPCTimerInterface* timerInterface = timerInterfaces[0]; + LogPrint("rpc", "queue run of timer %s in %i seconds (using %s)\n", name, nSeconds, timerInterface->Name()); + deadlineTimers.insert(std::make_pair(name, timerInterface->NewTimer(func, nSeconds))); +} + const CRPCTable tableRPC; |