Page MenuHomePhabricator

httpserver.cpp
No OneTemporary

httpserver.cpp

// Copyright (c) 2015-2016 The Bitcoin Core developers
// Copyright (c) 2018-2019 The Bitcoin developers
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
#include <httpserver.h>
#include <chainparamsbase.h>
#include <compat.h>
#include <config.h>
#include <logging.h>
#include <netbase.h>
#include <node/ui_interface.h>
#include <rpc/protocol.h> // For HTTP status codes
#include <shutdown.h>
#include <sync.h>
#include <util/strencodings.h>
#include <util/system.h>
#include <util/threadnames.h>
#include <util/translation.h>
#include <event2/buffer.h>
#include <event2/bufferevent.h>
#include <event2/keyvalq_struct.h>
#include <event2/thread.h>
#include <event2/util.h>
#include <support/events.h>
#ifdef EVENT__HAVE_NETINET_IN_H
#include <netinet/in.h>
#ifdef _XOPEN_SOURCE_EXTENDED
#include <arpa/inet.h>
#endif
#endif
#include <sys/stat.h>
#include <sys/types.h>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <deque>
#include <memory>
/** Maximum size of http request (request line + headers) */
static const size_t MAX_HEADERS_SIZE = 8192;
/**
* Maximum HTTP post body size. Twice the maximum block size is added to this
* value in practice.
*/
static const size_t MIN_SUPPORTED_BODY_SIZE = 0x02000000;
/** HTTP request work item */
class HTTPWorkItem final : public HTTPClosure {
public:
HTTPWorkItem(Config &_config, std::unique_ptr<HTTPRequest> _req,
const std::string &_path, const HTTPRequestHandler &_func)
: req(std::move(_req)), path(_path), func(_func), config(&_config) {}
void operator()() override { func(*config, req.get(), path); }
std::unique_ptr<HTTPRequest> req;
private:
std::string path;
HTTPRequestHandler func;
Config *config;
};
/**
* Simple work queue for distributing work over multiple threads.
* Work items are simply callable objects.
*/
template <typename WorkItem> class WorkQueue {
private:
/** Mutex protects entire object */
Mutex cs;
std::condition_variable cond;
std::deque<std::unique_ptr<WorkItem>> queue;
bool running;
size_t maxDepth;
public:
explicit WorkQueue(size_t _maxDepth) : running(true), maxDepth(_maxDepth) {}
/**
* Precondition: worker threads have all stopped (they have all been joined)
*/
~WorkQueue() {}
/** Enqueue a work item */
bool Enqueue(WorkItem *item) {
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() {
while (true) {
std::unique_ptr<WorkItem> i;
{
WAIT_LOCK(cs, lock);
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() {
LOCK(cs);
running = false;
cond.notify_all();
}
};
struct HTTPPathHandler {
HTTPPathHandler(std::string _prefix, bool _exactMatch,
HTTPRequestHandler _handler)
: prefix(_prefix), exactMatch(_exactMatch), handler(_handler) {}
std::string prefix;
bool exactMatch;
HTTPRequestHandler handler;
};
/** HTTP module state */
//! libevent event loop
static struct event_base *eventBase = nullptr;
//! HTTP server
static struct evhttp *eventHTTP = nullptr;
//! List of subnets to allow RPC connections from
static std::vector<CSubNet> rpc_allow_subnets;
//! Work queue for handling longer requests off the event loop thread
static WorkQueue<HTTPClosure> *workQueue = nullptr;
//! Handlers for (sub)paths
static std::vector<HTTPPathHandler> pathHandlers;
//! Bound listening sockets
static std::vector<evhttp_bound_socket *> boundSockets;
/** Check if a network address is allowed to access the HTTP server */
static bool ClientAllowed(const CNetAddr &netaddr) {
if (!netaddr.IsValid()) {
return false;
}
for (const CSubNet &subnet : rpc_allow_subnets) {
if (subnet.Match(netaddr)) {
return true;
}
}
return false;
}
/** Initialize ACL list for HTTP server */
static bool InitHTTPAllowList() {
rpc_allow_subnets.clear();
CNetAddr localv4;
CNetAddr localv6;
LookupHost("127.0.0.1", localv4, false);
LookupHost("::1", localv6, false);
// always allow IPv4 local subnet.
rpc_allow_subnets.push_back(CSubNet(localv4, 8));
// always allow IPv6 localhost.
rpc_allow_subnets.push_back(CSubNet(localv6));
for (const std::string &strAllow : gArgs.GetArgs("-rpcallowip")) {
CSubNet subnet;
LookupSubNet(strAllow, subnet);
if (!subnet.IsValid()) {
uiInterface.ThreadSafeMessageBox(
strprintf(
Untranslated("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;
}
/** HTTP request method as string - use for logging only */
std::string RequestMethodString(HTTPRequest::RequestMethod m) {
switch (m) {
case HTTPRequest::GET:
return "GET";
case HTTPRequest::POST:
return "POST";
case HTTPRequest::HEAD:
return "HEAD";
case HTTPRequest::PUT:
return "PUT";
case HTTPRequest::OPTIONS:
return "OPTIONS";
default:
return "unknown";
}
}
/** HTTP request callback */
static void http_request_cb(struct evhttp_request *req, void *arg) {
Config &config = *reinterpret_cast<Config *>(arg);
// 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);
}
}
}
auto hreq = std::make_unique<HTTPRequest>(req);
// Early address-based allow check
if (!ClientAllowed(hreq->GetPeer())) {
LogPrint(BCLog::HTTP,
"HTTP request from %s rejected: Client network is not allowed "
"RPC access\n",
hreq->GetPeer().ToString());
hreq->WriteReply(HTTP_FORBIDDEN);
return;
}
// Early reject unknown HTTP methods
if (hreq->GetRequestMethod() == HTTPRequest::UNKNOWN) {
LogPrint(BCLog::HTTP,
"HTTP request from %s rejected: Unknown HTTP request method\n",
hreq->GetPeer().ToString());
hreq->WriteReply(HTTP_BAD_METHOD);
return;
}
LogPrint(BCLog::HTTP, "Received a %s request for %s from %s\n",
RequestMethodString(hreq->GetRequestMethod()),
SanitizeString(hreq->GetURI(), SAFE_CHARS_URI).substr(0, 100),
hreq->GetPeer().ToString());
// 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(config, std::move(hreq), path, i->handler));
assert(workQueue);
if (workQueue->Enqueue(item.get())) {
/* if true, queue took ownership */
item.release();
} 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_SERVER_ERROR,
"Work queue depth exceeded");
}
} else {
hreq->WriteReply(HTTP_NOT_FOUND);
}
}
/** Callback to reject HTTP requests after shutdown. */
static void http_reject_request_cb(struct evhttp_request *req, void *) {
LogPrint(BCLog::HTTP, "Rejecting request while shutting down\n");
evhttp_send_error(req, HTTP_SERVUNAVAIL, nullptr);
}
/** Event dispatcher thread */
static bool ThreadHTTP(struct event_base *base) {
util::ThreadRename("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;
}
/** Bind HTTP server to specified addresses */
static bool HTTPBindAddresses(struct evhttp *http) {
int http_port = gArgs.GetArg("-rpcport", BaseParams().RPCPort());
std::vector<std::pair<std::string, uint16_t>> endpoints;
// Determine what addresses to bind to
if (!(gArgs.IsArgSet("-rpcallowip") && gArgs.IsArgSet("-rpcbind"))) {
// Default to loopback if not allowing external IPs.
endpoints.push_back(std::make_pair("::1", http_port));
endpoints.push_back(std::make_pair("127.0.0.1", http_port));
if (gArgs.IsArgSet("-rpcallowip")) {
LogPrintf("WARNING: option -rpcallowip was specified without "
"-rpcbind; this doesn't usually make sense\n");
}
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 = http_port;
std::string host;
SplitHostPort(strRPCBind, port, host);
endpoints.push_back(std::make_pair(host, port));
}
}
// 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) {
CNetAddr addr;
if (i->first.empty() ||
(LookupHost(i->first, addr, false) && addr.IsBindAny())) {
LogPrintf("WARNING: the RPC server is not safe to expose to "
"untrusted networks such as the public internet\n");
}
boundSockets.push_back(bind_handle);
} else {
LogPrintf("Binding RPC on address %s port %i failed.\n", i->first,
i->second);
}
}
return !boundSockets.empty();
}
/** Simple wrapper to set thread name and run work queue */
static void HTTPWorkQueueRun(WorkQueue<HTTPClosure> *queue, int worker_num) {
util::ThreadRename(strprintf("httpworker.%i", worker_num));
queue->Run();
}
/** libevent event log callback */
static void libevent_log_cb(int severity, const char *msg) {
#ifndef EVENT_LOG_WARN
// EVENT_LOG_WARN was added in 2.0.19; but before then _EVENT_LOG_WARN existed.
#define EVENT_LOG_WARN _EVENT_LOG_WARN
#endif
// Log warn messages and higher without debug category.
if (severity >= EVENT_LOG_WARN) {
LogPrintf("libevent: %s\n", msg);
} else {
LogPrint(BCLog::LIBEVENT, "libevent: %s\n", msg);
}
}
bool InitHTTPServer(Config &config) {
if (!InitHTTPAllowList()) {
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(
LogInstance().WillLogCategory(BCLog::LIBEVENT))) {
LogInstance().DisableCategory(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, MIN_SUPPORTED_BODY_SIZE +
2 * config.GetMaxBlockSize());
evhttp_set_gencb(http, http_request_cb, &config);
// Only POST and OPTIONS are supported, but we return HTTP 405 for the
// others
evhttp_set_allowed_methods(
http, EVHTTP_REQ_GET | EVHTTP_REQ_POST | EVHTTP_REQ_HEAD |
EVHTTP_REQ_PUT | EVHTTP_REQ_DELETE | EVHTTP_REQ_OPTIONS);
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;
}
bool UpdateHTTPServerLogging(bool enable) {
#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
}
static std::thread threadHTTP;
static std::vector<std::thread> g_thread_http_workers;
void StartHTTPServer() {
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);
threadHTTP = std::thread(ThreadHTTP, eventBase);
for (int i = 0; i < rpcThreads; i++) {
g_thread_http_workers.emplace_back(HTTPWorkQueueRun, workQueue, i);
}
}
void InterruptHTTPServer() {
LogPrint(BCLog::HTTP, "Interrupting HTTP server\n");
if (eventHTTP) {
// Reject requests on current connections
evhttp_set_gencb(eventHTTP, http_reject_request_cb, nullptr);
}
if (workQueue) {
workQueue->Interrupt();
}
}
void StopHTTPServer() {
LogPrint(BCLog::HTTP, "Stopping HTTP server\n");
if (workQueue) {
LogPrint(BCLog::HTTP, "Waiting for HTTP worker threads to exit\n");
for (auto &thread : g_thread_http_workers) {
thread.join();
}
g_thread_http_workers.clear();
delete workQueue;
workQueue = nullptr;
}
// Unlisten sockets, these are what make the event loop running, which means
// that after this and all connections are closed the event loop will quit.
for (evhttp_bound_socket *socket : boundSockets) {
evhttp_del_accept_socket(eventHTTP, socket);
}
boundSockets.clear();
if (eventBase) {
LogPrint(BCLog::HTTP, "Waiting for HTTP event thread to exit\n");
threadHTTP.join();
}
if (eventHTTP) {
evhttp_free(eventHTTP);
eventHTTP = nullptr;
}
if (eventBase) {
event_base_free(eventBase);
eventBase = nullptr;
}
LogPrint(BCLog::HTTP, "Stopped HTTP server\n");
}
struct event_base *EventBase() {
return eventBase;
}
static void httpevent_callback_fn(evutil_socket_t, short, void *data) {
// Static handler: simply call inner handler
HTTPEvent *self = static_cast<HTTPEvent *>(data);
self->handler();
if (self->deleteWhenTriggered) {
delete self;
}
}
HTTPEvent::HTTPEvent(struct event_base *base, bool _deleteWhenTriggered,
const std::function<void()> &_handler)
: deleteWhenTriggered(_deleteWhenTriggered), handler(_handler) {
ev = event_new(base, -1, 0, httpevent_callback_fn, this);
assert(ev);
}
HTTPEvent::~HTTPEvent() {
event_free(ev);
}
void HTTPEvent::trigger(struct timeval *tv) {
if (tv == nullptr) {
// Immediately trigger event in main thread.
event_active(ev, 0, 0);
} else {
// Trigger after timeval passed.
evtimer_add(ev, tv);
}
}
HTTPRequest::HTTPRequest(struct evhttp_request *_req, bool _replySent)
: req(_req), replySent(_replySent) {}
HTTPRequest::~HTTPRequest() {
if (!replySent) {
// Keep track of whether reply was sent to avoid request leaks
LogPrintf("%s: Unhandled request\n", __func__);
WriteReply(HTTP_INTERNAL_SERVER_ERROR, "Unhandled request");
}
// evhttpd cleans up the request, as long as a reply was sent.
}
std::pair<bool, std::string>
HTTPRequest::GetHeader(const std::string &hdr) const {
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, "");
}
}
std::string HTTPRequest::ReadBody() {
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,
* internal copying can be avoided in multi-segment buffers by using
* evbuffer_peek and an awkward loop. Though in that case, it'd be even
* better to not copy into an intermediate string but use a stream
* abstraction to consume the evbuffer on the fly in the parsing algorithm.
*/
const char *data = (const char *)evbuffer_pullup(buf, size);
// returns nullptr in case of empty buffer.
if (!data) {
return "";
}
std::string rv(data, size);
evbuffer_drain(buf, size);
return rv;
}
void HTTPRequest::WriteHeader(const std::string &hdr,
const std::string &value) {
struct evkeyvalq *headers = evhttp_request_get_output_headers(req);
assert(headers);
evhttp_add_header(headers, hdr.c_str(), value.c_str());
}
/**
* Closure sent to main thread to request a reply to be sent to a HTTP request.
* Replies must be sent in the main loop in the main http thread, this cannot be
* done from worker threads.
*/
void HTTPRequest::WriteReply(int nStatus, const std::string &strReply) {
assert(!replySent && req);
if (ShutdownRequested()) {
WriteHeader("Connection", "close");
}
// 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;
// transferred back to main thread.
req = nullptr;
}
CService HTTPRequest::GetPeer() const {
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;
}
std::string HTTPRequest::GetURI() const {
return evhttp_request_get_uri(req);
}
HTTPRequest::RequestMethod HTTPRequest::GetRequestMethod() const {
switch (evhttp_request_get_command(req)) {
case EVHTTP_REQ_GET:
return GET;
case EVHTTP_REQ_POST:
return POST;
case EVHTTP_REQ_HEAD:
return HEAD;
case EVHTTP_REQ_PUT:
return PUT;
case EVHTTP_REQ_OPTIONS:
return OPTIONS;
default:
return UNKNOWN;
}
}
void RegisterHTTPHandler(const std::string &prefix, bool exactMatch,
const HTTPRequestHandler &handler) {
LogPrint(BCLog::HTTP, "Registering HTTP handler for %s (exactmatch %d)\n",
prefix, exactMatch);
pathHandlers.push_back(HTTPPathHandler(prefix, exactMatch, handler));
}
void UnregisterHTTPHandler(const std::string &prefix, bool exactMatch) {
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);
}
}

File Metadata

Mime Type
text/x-c++
Expires
Sat, Nov 23, 09:55 (1 d, 11 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
4528099
Default Alt Text
httpserver.cpp (22 KB)

Event Timeline