{

HTTPWorkItem(std::unique_ptr<HTTPRequest> _req, const std::string &_path, const HTTPRequestHandler& _func):

req(std::move(_req)), path(_path), func(_func)

{

}

void operator()() override

{

func(req.get(), path);

}

std::unique_ptr<HTTPRequest> req;

std::string path;

HTTPRequestHandler func;

{

/** Mutex protects entire object */

std::mutex cs;

std::condition_variable cond;

std::deque<std::unique_ptr<WorkItem>> queue;

bool running;

size_t maxDepth;

int numThreads;

/** RAII object to keep track of number of running worker threads */

class ThreadCounter

{

public:

WorkQueue &wq;

explicit ThreadCounter(WorkQueue &w): wq(w)

{

std::lock_guard<std::mutex> lock(wq.cs);

wq.numThreads += 1;

}

~ThreadCounter()

{

std::lock_guard<std::mutex> lock(wq.cs);

wq.numThreads -= 1;

wq.cond.notify_all();

}

};

explicit WorkQueue(size_t _maxDepth) : running(true),

maxDepth(_maxDepth),

numThreads(0)

{

}

/** Precondition: worker threads have all stopped

~WorkQueue()

{

}

/** Enqueue a work item */

bool Enqueue(WorkItem* item)

{

std::unique_lock<std::mutex> lock(cs);

if (queue.size() >= maxDepth) {

return false;

}

queue.emplace_back(std::unique_ptr<WorkItem>(item));

cond.notify_one();

return true;

}

/** Thread function */

void Run()

{

ThreadCounter count(*this);

while (true) {

std::unique_ptr<WorkItem> i;

{

std::unique_lock<std::mutex> lock(cs);

while (running && queue.empty())

cond.wait(lock);

if (!running)

break;

i = std::move(queue.front());

queue.pop_front();

}

(*i)();

}

}

/** Interrupt and exit loops */

void Interrupt()

{

std::unique_lock<std::mutex> lock(cs);

running = false;

cond.notify_all();

}

/** Wait for worker threads to exit */

void WaitExit()

{

std::unique_lock<std::mutex> lock(cs);

while (numThreads > 0)

cond.wait(lock);

}

{

HTTPPathHandler() {}

HTTPPathHandler(std::string _prefix, bool _exactMatch, HTTPRequestHandler _handler):

prefix(_prefix), exactMatch(_exactMatch), handler(_handler)

{

}

std::string prefix;

bool exactMatch;

HTTPRequestHandler handler;

{

if (!netaddr.IsValid())

return false;

for(const CSubNet& subnet : rpc_allow_subnets)

if (subnet.Match(netaddr))

return true;

return false;

{

rpc_allow_subnets.clear();

CNetAddr localv4;

CNetAddr localv6;

LookupHost("127.0.0.1", localv4, false);

LookupHost("::1", localv6, false);

rpc_allow_subnets.push_back(CSubNet(localv4, 8)); // always allow IPv4 local subnet

rpc_allow_subnets.push_back(CSubNet(localv6)); // always allow IPv6 localhost

for (const std::string& strAllow : gArgs.GetArgs("-rpcallowip")) {

CSubNet subnet;

LookupSubNet(strAllow.c_str(), subnet);

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);

return false;

}

rpc_allow_subnets.push_back(subnet);

}

std::string strAllowed;

for (const CSubNet& subnet : rpc_allow_subnets)

strAllowed += subnet.ToString() + " ";

LogPrint(BCLog::HTTP, "Allowing HTTP connections from: %s\n", strAllowed);

return true;

{

switch (m) {

case HTTPRequest::GET:

return "GET";

break;

case HTTPRequest::POST:

return "POST";

break;

case HTTPRequest::HEAD:

return "HEAD";

break;

case HTTPRequest::PUT:

return "PUT";

break;

default:

return "unknown";

}

{

// Disable reading to work around a libevent bug, fixed in 2.2.0.

if (event_get_version_number() >= 0x02010600 && event_get_version_number() < 0x02020001) {

evhttp_connection* conn = evhttp_request_get_connection(req);

if (conn) {

bufferevent* bev = evhttp_connection_get_bufferevent(conn);

if (bev) {

bufferevent_disable(bev, EV_READ);

}

}

}

std::unique_ptr<HTTPRequest> hreq(new HTTPRequest(req));

LogPrint(BCLog::HTTP, "Received a %s request for %s from %s\n",

RequestMethodString(hreq->GetRequestMethod()), hreq->GetURI(), hreq->GetPeer().ToString());

// Early address-based allow check

if (!ClientAllowed(hreq->GetPeer())) {

hreq->WriteReply(HTTP_FORBIDDEN);

return;

}

// Early reject unknown HTTP methods

if (hreq->GetRequestMethod() == HTTPRequest::UNKNOWN) {

hreq->WriteReply(HTTP_BADMETHOD);

return;

}

// Find registered handler for prefix

std::string strURI = hreq->GetURI();

std::string path;

std::vector<HTTPPathHandler>::const_iterator i = pathHandlers.begin();

std::vector<HTTPPathHandler>::const_iterator iend = pathHandlers.end();

for (; i != iend; ++i) {

bool match = false;

if (i->exactMatch)

match = (strURI == i->prefix);

else

match = (strURI.substr(0, i->prefix.size()) == i->prefix);

if (match) {

path = strURI.substr(i->prefix.size());

break;

}

}

// Dispatch to worker thread

if (i != iend) {

std::unique_ptr<HTTPWorkItem> item(new HTTPWorkItem(std::move(hreq), path, i->handler));

assert(workQueue);

if (workQueue->Enqueue(item.get()))

item.release(); /* if true, queue took ownership */

else {

LogPrintf("WARNING: request rejected because http work queue depth exceeded, it can be increased with the -rpcworkqueue= setting\n");

item->req->WriteReply(HTTP_INTERNAL, "Work queue depth exceeded");

}

} else {

hreq->WriteReply(HTTP_NOTFOUND);

}

{

LogPrint(BCLog::HTTP, "Rejecting request while shutting down\n");

evhttp_send_error(req, HTTP_SERVUNAVAIL, nullptr);

{

RenameThread("bitcoin-http");

LogPrint(BCLog::HTTP, "Entering http event loop\n");

event_base_dispatch(base);

// Event loop will be interrupted by InterruptHTTPServer()

LogPrint(BCLog::HTTP, "Exited http event loop\n");

return event_base_got_break(base) == 0;

{

int defaultPort = gArgs.GetArg("-rpcport", BaseParams().RPCPort());

std::vector<std::pair<std::string, uint16_t> > endpoints;

// Determine what addresses to bind to

if (!gArgs.IsArgSet("-rpcallowip")) { // Default to loopback if not allowing external IPs

endpoints.push_back(std::make_pair("::1", defaultPort));

endpoints.push_back(std::make_pair("127.0.0.1", defaultPort));

if (gArgs.IsArgSet("-rpcbind")) {

LogPrintf("WARNING: option -rpcbind was ignored because -rpcallowip was not specified, refusing to allow everyone to connect\n");

}

} else if (gArgs.IsArgSet("-rpcbind")) { // Specific bind address

for (const std::string& strRPCBind : gArgs.GetArgs("-rpcbind")) {

int port = defaultPort;

std::string host;

SplitHostPort(strRPCBind, port, host);

endpoints.push_back(std::make_pair(host, port));

}

} else { // No specific bind address specified, bind to any

endpoints.push_back(std::make_pair("::", defaultPort));

endpoints.push_back(std::make_pair("0.0.0.0", defaultPort));

}

// Bind addresses

for (std::vector<std::pair<std::string, uint16_t> >::iterator i = endpoints.begin(); i != endpoints.end(); ++i) {

LogPrint(BCLog::HTTP, "Binding RPC on address %s port %i\n", i->first, i->second);

evhttp_bound_socket *bind_handle = evhttp_bind_socket_with_handle(http, i->first.empty() ? nullptr : i->first.c_str(), i->second);

if (bind_handle) {

boundSockets.push_back(bind_handle);

} else {

LogPrintf("Binding RPC on address %s port %i failed.\n", i->first, i->second);

}

}

return !boundSockets.empty();

{

RenameThread("bitcoin-httpworker");

queue->Run();

{

#ifndef EVENT_LOG_WARN

# define EVENT_LOG_WARN _EVENT_LOG_WARN

#endif

if (severity >= EVENT_LOG_WARN) // Log warn messages and higher without debug category

LogPrintf("libevent: %s\n", msg);

else

LogPrint(BCLog::LIBEVENT, "libevent: %s\n", msg);

{

if (!InitHTTPAllowList())

return false;

if (gArgs.GetBoolArg("-rpcssl", false)) {

uiInterface.ThreadSafeMessageBox(

"SSL mode for RPC (-rpcssl) is no longer supported.",

"", CClientUIInterface::MSG_ERROR);

return false;

}

// Redirect libevent's logging to our own log

event_set_log_callback(&libevent_log_cb);

// Update libevent's log handling. Returns false if our version of

// libevent doesn't support debug logging, in which case we should

// clear the BCLog::LIBEVENT flag.

if (!UpdateHTTPServerLogging(logCategories & BCLog::LIBEVENT)) {

logCategories &= ~BCLog::LIBEVENT;

}

#ifdef WIN32

evthread_use_windows_threads();

#else

evthread_use_pthreads();

#endif

raii_event_base base_ctr = obtain_event_base();

/* Create a new evhttp object to handle requests. */

raii_evhttp http_ctr = obtain_evhttp(base_ctr.get());

struct evhttp* http = http_ctr.get();

if (!http) {

LogPrintf("couldn't create evhttp. Exiting.\n");

return false;

}

evhttp_set_timeout(http, gArgs.GetArg("-rpcservertimeout", DEFAULT_HTTP_SERVER_TIMEOUT));

evhttp_set_max_headers_size(http, MAX_HEADERS_SIZE);

evhttp_set_max_body_size(http, MAX_SIZE);

evhttp_set_gencb(http, http_request_cb, nullptr);

if (!HTTPBindAddresses(http)) {

LogPrintf("Unable to bind any endpoint for RPC server\n");

return false;

}

LogPrint(BCLog::HTTP, "Initialized HTTP server\n");

int workQueueDepth = std::max((long)gArgs.GetArg("-rpcworkqueue", DEFAULT_HTTP_WORKQUEUE), 1L);

LogPrintf("HTTP: creating work queue of depth %d\n", workQueueDepth);

workQueue = new WorkQueue<HTTPClosure>(workQueueDepth);

// transfer ownership to eventBase/HTTP via .release()

eventBase = base_ctr.release();

eventHTTP = http_ctr.release();

return true;

#if LIBEVENT_VERSION_NUMBER >= 0x02010100

if (enable) {

event_enable_debug_logging(EVENT_DBG_ALL);

} else {

event_enable_debug_logging(EVENT_DBG_NONE);

}

return true;

#else

// Can't update libevent logging if version < 02010100

return false;

#endif

{

LogPrint(BCLog::HTTP, "Starting HTTP server\n");

int rpcThreads = std::max((long)gArgs.GetArg("-rpcthreads", DEFAULT_HTTP_THREADS), 1L);

LogPrintf("HTTP: starting %d worker threads\n", rpcThreads);

std::packaged_task<bool(event_base*, evhttp*)> task(ThreadHTTP);

threadResult = task.get_future();

threadHTTP = std::thread(std::move(task), eventBase, eventHTTP);

for (int i = 0; i < rpcThreads; i++) {

std::thread rpc_worker(HTTPWorkQueueRun, workQueue);

rpc_worker.detach();

}

return true;

{

LogPrint(BCLog::HTTP, "Interrupting HTTP server\n");

if (eventHTTP) {

// Unlisten sockets

for (evhttp_bound_socket *socket : boundSockets) {

evhttp_del_accept_socket(eventHTTP, socket);

}

// Reject requests on current connections

evhttp_set_gencb(eventHTTP, http_reject_request_cb, nullptr);

}

if (workQueue)

workQueue->Interrupt();

{

LogPrint(BCLog::HTTP, "Stopping HTTP server\n");

if (workQueue) {

LogPrint(BCLog::HTTP, "Waiting for HTTP worker threads to exit\n");

workQueue->WaitExit();

delete workQueue;

workQueue = nullptr;

}

if (eventBase) {

LogPrint(BCLog::HTTP, "Waiting for HTTP event thread to exit\n");

// Exit the event loop as soon as there are no active events.

event_base_loopexit(eventBase, nullptr);

// Give event loop a few seconds to exit (to send back last RPC responses), then break it

// Before this was solved with event_base_loopexit, but that didn't work as expected in

// at least libevent 2.0.21 and always introduced a delay. In libevent

// master that appears to be solved, so in the future that solution

// could be used again (if desirable).

// (see discussion in https://github.com/bitcoin/bitcoin/pull/6990)

if (threadResult.valid() && threadResult.wait_for(std::chrono::milliseconds(2000)) == std::future_status::timeout) {

LogPrintf("HTTP event loop did not exit within allotted time, sending loopbreak\n");

event_base_loopbreak(eventBase);

}

threadHTTP.join();

}

if (eventHTTP) {

evhttp_free(eventHTTP);

eventHTTP = nullptr;

}

if (eventBase) {

event_base_free(eventBase);

eventBase = nullptr;

}

LogPrint(BCLog::HTTP, "Stopped HTTP server\n");

{

// Static handler: simply call inner handler

HTTPEvent *self = ((HTTPEvent*)data);

self->handler();

if (self->deleteWhenTriggered)

delete self;

deleteWhenTriggered(_deleteWhenTriggered), handler(_handler)

{

ev = event_new(base, -1, 0, httpevent_callback_fn, this);

assert(ev);

{

if (tv == nullptr)

event_active(ev, 0, 0); // immediately trigger event in main thread

else

evtimer_add(ev, tv); // trigger after timeval passed

{

const struct evkeyvalq* headers = evhttp_request_get_input_headers(req);

assert(headers);

const char* val = evhttp_find_header(headers, hdr.c_str());

if (val)

return std::make_pair(true, val);

else

return std::make_pair(false, "");

{

struct evbuffer* buf = evhttp_request_get_input_buffer(req);

if (!buf)

return "";

size_t size = evbuffer_get_length(buf);

/** Trivial implementation: if this is ever a performance bottleneck,

const char* data = (const char*)evbuffer_pullup(buf, size);

if (!data) // returns nullptr in case of empty buffer

return "";

std::string rv(data, size);

evbuffer_drain(buf, size);

return rv;

{

struct evkeyvalq* headers = evhttp_request_get_output_headers(req);

assert(headers);

evhttp_add_header(headers, hdr.c_str(), value.c_str());

{

assert(!replySent && req);

// Send event to main http thread to send reply message

struct evbuffer* evb = evhttp_request_get_output_buffer(req);

assert(evb);

evbuffer_add(evb, strReply.data(), strReply.size());

auto req_copy = req;

HTTPEvent* ev = new HTTPEvent(eventBase, true, [req_copy, nStatus]{

evhttp_send_reply(req_copy, nStatus, nullptr, nullptr);

// Re-enable reading from the socket. This is the second part of the libevent

// workaround above.

if (event_get_version_number() >= 0x02010600 && event_get_version_number() < 0x02020001) {

evhttp_connection* conn = evhttp_request_get_connection(req_copy);

if (conn) {

bufferevent* bev = evhttp_connection_get_bufferevent(conn);

if (bev) {

bufferevent_enable(bev, EV_READ | EV_WRITE);

}

}

}

});

ev->trigger(nullptr);

replySent = true;

req = nullptr; // transferred back to main thread

{

evhttp_connection* con = evhttp_request_get_connection(req);

CService peer;

if (con) {

// evhttp retains ownership over returned address string

const char* address = "";

uint16_t port = 0;

evhttp_connection_get_peer(con, (char**)&address, &port);

peer = LookupNumeric(address, port);

}

return peer;

{

switch (evhttp_request_get_command(req)) {

case EVHTTP_REQ_GET:

return GET;

break;

case EVHTTP_REQ_POST:

return POST;

break;

case EVHTTP_REQ_HEAD:

return HEAD;

break;

case EVHTTP_REQ_PUT:

return PUT;

break;

default:

return UNKNOWN;

break;

}

{

LogPrint(BCLog::HTTP, "Registering HTTP handler for %s (exactmatch %d)\n", prefix, exactMatch);

pathHandlers.push_back(HTTPPathHandler(prefix, exactMatch, handler));

{

std::vector<HTTPPathHandler>::iterator i = pathHandlers.begin();

std::vector<HTTPPathHandler>::iterator iend = pathHandlers.end();

for (; i != iend; ++i)

if (i->prefix == prefix && i->exactMatch == exactMatch)

break;

if (i != iend)

{

LogPrint(BCLog::HTTP, "Unregistering HTTP handler for %s (exactmatch %d)\n", prefix, exactMatch);

pathHandlers.erase(i);

}

std::string res;

if (!urlEncoded.empty()) {

char *decoded = evhttp_uridecode(urlEncoded.c_str(), false, nullptr);

if (decoded) {

res = std::string(decoded);

free(decoded);

}

}

return res;