/* * This file is part of the Flowee project * Copyright (c) 2015 The Bitcoin Core developers * Copyright (c) 2024 Tom Zander * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation, either version 3 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ #include "httpserver.h" #include #include "chainparamsbase.h" #include #include "compat.h" #include "util.h" #include "netbase.h" #include "rpcprotocol.h" // For HTTP status codes #include "sync.h" #include "UiInterface.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #ifdef EVENT__HAVE_NETINET_IN_H #include #ifdef _XOPEN_SOURCE_EXTENDED #include #endif #endif #include // for to_lower() #include #include /** Maximum size of http request (request line + headers) */ static constexpr size_t MAX_HEADERS_SIZE = 8192; /** * Maximum HTTP post body size. Twice the maximum block size is added to this * value in practice. */ static constexpr size_t MIN_SUPPORTED_BODY_SIZE = 0x02000000; /** HTTP request work item */ class HTTPWorkItem : public HTTPClosure { public: HTTPWorkItem(HTTPRequest* req, const std::string &path, const HTTPRequestHandler& func): req(req), path(path), func(func) { } void operator()() { func(req.get(), path); } boost::scoped_ptr req; private: std::string path; HTTPRequestHandler func; }; /** Simple work queue for distributing work over multiple threads. * Work items are simply callable objects. */ template class WorkQueue { private: /** Mutex protects entire object */ CWaitableCriticalSection cs; CConditionVariable cond; /* XXX in C++11 we can use std::unique_ptr here and avoid manual cleanup */ std::deque queue; bool running; size_t maxDepth; int numThreads; /** RAII object to keep track of number of running worker threads */ class ThreadCounter { public: WorkQueue &wq; ThreadCounter(WorkQueue &w): wq(w) { boost::lock_guard lock(wq.cs); wq.numThreads += 1; } ~ThreadCounter() { boost::lock_guard lock(wq.cs); wq.numThreads -= 1; wq.cond.notify_all(); } }; public: WorkQueue(size_t maxDepth) : running(true), maxDepth(maxDepth), numThreads(0) { } /*( Precondition: worker threads have all stopped * (call WaitExit) */ ~WorkQueue() { while (!queue.empty()) { delete queue.front(); queue.pop_front(); } } /** Enqueue a work item */ bool Enqueue(WorkItem* item) { boost::unique_lock lock(cs); if (queue.size() >= maxDepth) { return false; } queue.push_back(item); cond.notify_one(); return true; } /** Thread function */ void Run() { ThreadCounter count(*this); while (running) { WorkItem* i = 0; { boost::unique_lock lock(cs); while (running && queue.empty()) cond.wait(lock); if (!running) break; i = queue.front(); queue.pop_front(); } (*i)(); delete i; } } /** Interrupt and exit loops */ void Interrupt() { boost::unique_lock lock(cs); running = false; cond.notify_all(); } /** Wait for worker threads to exit */ void WaitExit() { boost::unique_lock lock(cs); while (numThreads > 0) cond.wait(lock); } /** Return current depth of queue */ size_t Depth() { boost::unique_lock lock(cs); return queue.size(); } }; struct HTTPPathHandler { 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 = 0; //! HTTP server struct evhttp* eventHTTP = 0; //! List of subnets to allow RPC connections from static std::vector rpc_allow_subnets; //! Work queue for handling longer requests off the event loop thread static WorkQueue* workQueue = 0; //! Handlers for (sub)paths std::vector pathHandlers; //! Bound listening sockets std::vector boundSockets; /** Check if a network address is allowed to access the HTTP server */ static bool ClientAllowed(const CNetAddr& netaddr) { if (!netaddr.IsValid()) return false; BOOST_FOREACH (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(); 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 std::vector& vAllow = mapMultiArgs["-rpcallowip"]; BOOST_FOREACH (std::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); return false; } rpc_allow_subnets.push_back(subnet); } } std::string strAllowed; BOOST_FOREACH (const CSubNet& subnet, rpc_allow_subnets) strAllowed += subnet.ToString() + " "; logCritical(Log::HTTP) << "Allowing HTTP connections from:" << strAllowed; return true; } /** HTTP request method as string - use for logging only */ static std::string RequestMethodString(HTTPRequest::RequestMethod m) { 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"; } } /** HTTP request callback */ static void http_request_cb(struct evhttp_request* req, void* arg) { std::unique_ptr hreq(new HTTPRequest(req)); logInfo(Log::HTTP) << "Received a" << RequestMethodString(hreq->GetRequestMethod()) << "request for" << hreq->GetURI() << "from" << 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::const_iterator i = pathHandlers.begin(); std::vector::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 item(new HTTPWorkItem(hreq.release(), path, i->handler)); assert(workQueue); if (workQueue->Enqueue(item.get())) item.release(); /* if true, queue took ownership */ else item->req->WriteReply(HTTP_INTERNAL, "Work queue depth exceeded"); } else { hreq->WriteReply(HTTP_NOTFOUND); } } /** Callback to reject HTTP requests after shutdown. */ static void http_reject_request_cb(struct evhttp_request* req, void*) { logDebug(Log::HTTP) << "Rejecting request while shutting down"; evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL); } /** Event dispatcher thread */ static void ThreadHTTP(struct event_base* base, struct evhttp* http) { RenameThread("bitcoin-http"); logDebug(Log::HTTP) << "Entering http event loop"; event_base_dispatch(base); // Event loop will be interrupted by InterruptHTTPServer() logDebug(Log::HTTP) << "Exited http event loop"; } /** Bind HTTP server to specified addresses */ static bool HTTPBindAddresses(struct evhttp* http) { int defaultPort = GetArg("-rpcport", BaseParams().RPCPort()); std::vector > endpoints; // Determine what addresses to bind to if (!mapArgs.count("-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 (mapArgs.count("-rpcbind")) { logCritical(Log::HTTP) << "WARNING: option -rpcbind was ignored because -rpcallowip was not specified, refusing to allow everyone to connect"; } } else if (mapArgs.count("-rpcbind")) { // Specific bind address const std::vector& vbind = mapMultiArgs["-rpcbind"]; for (std::vector::const_iterator i = vbind.begin(); i != vbind.end(); ++i) { uint16_t port = defaultPort; std::string host; SplitHostPort(*i, 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 >::iterator i = endpoints.begin(); i != endpoints.end(); ++i) { logInfo(Log::HTTP) << "Binding RPC on address" << i->first << "port" << i->second; evhttp_bound_socket *bind_handle = evhttp_bind_socket_with_handle(http, i->first.empty() ? NULL : i->first.c_str(), i->second); if (bind_handle) { boundSockets.push_back(bind_handle); } else { logFatal(Log::HTTP) <<"Binding RPC on address" << i->first << "port" << i->second << "failed"; } } return !boundSockets.empty(); } /** Simple wrapper to set thread name and run work queue */ static void HTTPWorkQueueRun(WorkQueue* queue) { RenameThread("httpworker"); 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 // unfortunately some methods on libEvent (notably evhttp_add_header) // adds a linefeed to the message. // We copy the message on stack and strip off a linefeed if there is one. char buf[170]; const int stringLen = strnlen(msg, sizeof(buf)); memcpy(buf, msg, stringLen); buf[sizeof(buf) - 1] = 0; // ensure terminating zero if ('\n' == buf[stringLen - 1]) buf[stringLen - 1] = 0; // remove trailing linefeed. switch (severity) { case EVENT_LOG_DEBUG: logDebug(Log::LibEvent) << buf; break; case EVENT_LOG_MSG: logInfo(Log::LibEvent) << buf; break; case EVENT_LOG_WARN: logWarning(Log::LibEvent) << buf; break; case EVENT_LOG_ERR: logCritical(Log::LibEvent) << buf; break; } } bool InitHTTPServer() { struct evhttp* http = 0; struct event_base* base = 0; if (!InitHTTPAllowList()) return false; // Redirect libevent's logging to our own log event_set_log_callback(&libevent_log_cb); #if LIBEVENT_VERSION_NUMBER >= 0x02010100 if (Log::Manager::instance()->isEnabled(Log::LibEvent, Log::DebugLevel)) event_enable_debug_logging(EVENT_DBG_ALL); else event_enable_debug_logging(EVENT_DBG_NONE); #endif #ifdef WIN32 evthread_use_windows_threads(); #else evthread_use_pthreads(); #endif base = event_base_new(); // XXX RAII if (!base) { logCritical(Log::LibEvent) << "Couldn't create an event_base: exiting"; return false; } /* Create a new evhttp object to handle requests. */ http = evhttp_new(base); // XXX RAII if (!http) { logCritical(Log::LibEvent) << "couldn't create evhttp. Exiting."; event_base_free(base); return false; } evhttp_set_timeout(http, GetArg("-rpcservertimeout", Settings::DefaultHttpServerTimeout)); evhttp_set_max_headers_size(http, MAX_HEADERS_SIZE); evhttp_set_max_body_size(http, MIN_SUPPORTED_BODY_SIZE + 2 * Policy::blockSizeAcceptLimit()); evhttp_set_gencb(http, http_request_cb, NULL); if (!HTTPBindAddresses(http)) { logCritical(Log::LibEvent) << "Unable to bind any endpoint for RPC server"; evhttp_free(http); event_base_free(base); return false; } logCritical(Log::HTTP) << "Initialized HTTP server"; int workQueueDepth = std::max((long)GetArg("-rpcworkqueue", Settings::DefaultHttpWorkQueue), 1L); logInfo(Log::HTTP) << "creating work queue of depth" << workQueueDepth; workQueue = new WorkQueue(workQueueDepth); eventBase = base; eventHTTP = http; return true; } boost::thread threadHTTP; void StartHTTPServer() { logInfo(Log::HTTP) << "Starting HTTP server"; int rpcThreads = std::max((long)GetArg("-rpcthreads", Settings::DefaultHttpThreads), 1L); logInfo(Log::HTTP) << "starting" << rpcThreads << "worker threads"; threadHTTP = boost::thread(std::bind(&ThreadHTTP, eventBase, eventHTTP)); for (int i = 0; i < rpcThreads; i++) boost::thread(std::bind(&HTTPWorkQueueRun, workQueue)); } void InterruptHTTPServer() { if (eventHTTP) { logCritical(Log::HTTP) << "Interrupting HTTP server"; // Unlisten sockets BOOST_FOREACH (evhttp_bound_socket *socket, boundSockets) { evhttp_del_accept_socket(eventHTTP, socket); } // Reject requests on current connections evhttp_set_gencb(eventHTTP, http_reject_request_cb, NULL); } if (workQueue) workQueue->Interrupt(); } void StopHTTPServer() { bool stopped = false; if (workQueue) { stopped = true; logCritical(Log::HTTP) << "Stopping HTTP server"; logInfo(Log::HTTP) << "Waiting for HTTP worker threads to exit"; workQueue->WaitExit(); delete workQueue; } if (eventBase) { stopped = true; logInfo(Log::HTTP) << "Waiting for HTTP event thread to exit"; // 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 BOOST_VERSION >= 105000 if (!threadHTTP.try_join_for(boost::chrono::milliseconds(2000))) { #else if (!threadHTTP.timed_join(boost::posix_time::milliseconds(2000))) { #endif logCritical(Log::HTTP) << "HTTP event loop did not exit within allotted time, sending loopbreak"; event_base_loopbreak(eventBase); threadHTTP.join(); } } if (eventHTTP) { stopped = true; evhttp_free(eventHTTP); eventHTTP = 0; } if (eventBase) { stopped = true; event_base_free(eventBase); eventBase = 0; } if (stopped) logCritical(Log::HTTP) << "Stopped HTTP server"; } 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 = ((HTTPEvent*)data); self->handler(); if (self->deleteWhenTriggered) delete self; } HTTPEvent::HTTPEvent(struct event_base* base, bool deleteWhenTriggered, const boost::function& 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 == NULL) event_active(ev, 0, 0); // immediately trigger event in main thread else evtimer_add(ev, tv); // trigger after timeval passed } HTTPRequest::HTTPRequest(struct evhttp_request* req) : req(req), replySent(false) { } HTTPRequest::~HTTPRequest() { if (!replySent) { // Keep track of whether reply was sent to avoid request leaks logDebug(Log::HTTP) << "Unhandled request"; WriteReply(HTTP_INTERNAL, "Unhandled request"); } // evhttpd cleans up the request, as long as a reply was sent. } std::pair HTTPRequest::GetHeader(const std::string& hdr) { 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); if (!data) // returns NULL in case of empty buffer 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); // 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()); HTTPEvent* ev = new HTTPEvent(eventBase, true, std::bind(evhttp_send_reply, req, nStatus, (const char*)NULL, (struct evbuffer *)NULL)); ev->trigger(0); replySent = true; req = 0; // transferred back to main thread } CService HTTPRequest::GetPeer() { 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 = CService(address, port); } return peer; } std::string HTTPRequest::GetURI() { return evhttp_request_get_uri(req); } HTTPRequest::RequestMethod HTTPRequest::GetRequestMethod() { 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; } } void RegisterHTTPHandler(const std::string &prefix, bool exactMatch, const HTTPRequestHandler &handler) { logInfo(Log::HTTP) << "Registering HTTP handler for" << prefix << "exactmatch:" << exactMatch; pathHandlers.push_back(HTTPPathHandler(prefix, exactMatch, handler)); } void UnregisterHTTPHandler(const std::string &prefix, bool exactMatch) { std::vector::iterator i = pathHandlers.begin(); std::vector::iterator iend = pathHandlers.end(); for (; i != iend; ++i) if (i->prefix == prefix && i->exactMatch == exactMatch) break; if (i != iend) { logInfo(Log::HTTP) << "Unregistering HTTP handler for" << prefix << "exactmatch:" << exactMatch; pathHandlers.erase(i); } }