httpserver.cpp raw
1 // Copyright (c) 2015-2022 The Limenka developers
2 // Distributed under the MIT software license, see the accompanying
3 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
4
5 #include <httpserver.h>
6
7 #include <chainparamsbase.h>
8 #include <common/args.h>
9 #include <common/messages.h>
10 #include <compat/compat.h>
11 #include <logging.h>
12 #include <netbase.h>
13 #include <node/interface_ui.h>
14 #include <rpc/protocol.h> // For HTTP status codes
15 #include <sync.h>
16 #include <util/check.h>
17 #include <util/signalinterrupt.h>
18 #include <util/strencodings.h>
19 #include <util/threadnames.h>
20 #include <util/translation.h>
21
22 #include <condition_variable>
23 #include <cstdio>
24 #include <cstdlib>
25 #include <deque>
26 #include <memory>
27 #include <optional>
28 #include <span>
29 #include <string>
30 #include <unordered_map>
31
32 #include <sys/types.h>
33 #include <sys/stat.h>
34
35 #include <event2/buffer.h>
36 #include <event2/bufferevent.h>
37 #include <event2/http.h>
38 #include <event2/http_struct.h>
39 #include <event2/keyvalq_struct.h>
40 #include <event2/thread.h>
41 #include <event2/util.h>
42
43 #include <support/events.h>
44
45 using common::InvalidPortErrMsg;
46
47 /** Maximum size of http request (request line + headers) */
48 static const size_t MAX_HEADERS_SIZE = 8192;
49
50 /** HTTP request work item */
51 class HTTPWorkItem final : public HTTPClosure
52 {
53 public:
54 HTTPWorkItem(std::unique_ptr<HTTPRequest> _req, const std::string &_path, const HTTPRequestHandler& _func):
55 req(std::move(_req)), path(_path), func(_func)
56 {
57 }
58 void operator()() override
59 {
60 func(req.get(), path);
61 }
62
63 std::unique_ptr<HTTPRequest> req;
64
65 private:
66 std::string path;
67 HTTPRequestHandler func;
68 };
69
70 /** Simple work queue for distributing work over multiple threads.
71 * Work items are simply callable objects.
72 */
73 template <typename WorkItem>
74 class WorkQueue
75 {
76 private:
77 Mutex cs;
78 std::condition_variable cond GUARDED_BY(cs);
79 std::deque<std::unique_ptr<WorkItem>> queue GUARDED_BY(cs);
80 bool running GUARDED_BY(cs){true};
81 const size_t maxDepth;
82
83 public:
84 explicit WorkQueue(size_t _maxDepth) : maxDepth(_maxDepth)
85 {
86 }
87 /** Precondition: worker threads have all stopped (they have been joined).
88 */
89 ~WorkQueue() = default;
90 /** Enqueue a work item */
91 bool Enqueue(WorkItem* item) EXCLUSIVE_LOCKS_REQUIRED(!cs)
92 {
93 LOCK(cs);
94 if (!running || queue.size() >= maxDepth) {
95 return false;
96 }
97 queue.emplace_back(std::unique_ptr<WorkItem>(item));
98 cond.notify_one();
99 return true;
100 }
101 /** Thread function */
102 void Run() EXCLUSIVE_LOCKS_REQUIRED(!cs)
103 {
104 while (true) {
105 std::unique_ptr<WorkItem> i;
106 {
107 WAIT_LOCK(cs, lock);
108 while (running && queue.empty())
109 cond.wait(lock);
110 if (!running && queue.empty())
111 break;
112 i = std::move(queue.front());
113 queue.pop_front();
114 }
115 (*i)();
116 }
117 }
118 /** Interrupt and exit loops */
119 void Interrupt() EXCLUSIVE_LOCKS_REQUIRED(!cs)
120 {
121 LOCK(cs);
122 running = false;
123 cond.notify_all();
124 }
125 };
126
127 struct HTTPPathHandler
128 {
129 HTTPPathHandler(std::string _prefix, bool _exactMatch, HTTPRequestHandler _handler):
130 prefix(_prefix), exactMatch(_exactMatch), handler(_handler)
131 {
132 }
133 std::string prefix;
134 bool exactMatch;
135 HTTPRequestHandler handler;
136 };
137
138 /** HTTP module state */
139
140 //! libevent event loop
141 static struct event_base* eventBase = nullptr;
142 //! HTTP server
143 static struct evhttp* eventHTTP = nullptr;
144 //! List of subnets to allow RPC connections from
145 static std::vector<CSubNet> rpc_allow_subnets;
146 //! Work queue for handling longer requests off the event loop thread
147 static std::unique_ptr<WorkQueue<HTTPClosure>> g_work_queue{nullptr};
148 //! Handlers for (sub)paths
149 static GlobalMutex g_httppathhandlers_mutex;
150 static std::vector<HTTPPathHandler> pathHandlers GUARDED_BY(g_httppathhandlers_mutex);
151 //! Bound listening sockets
152 static std::vector<evhttp_bound_socket *> boundSockets;
153
154 /**
155 * @brief Helps keep track of open `evhttp_connection`s with active `evhttp_requests`
156 *
157 */
158 class HTTPRequestTracker
159 {
160 private:
161 mutable Mutex m_mutex;
162 mutable std::condition_variable m_cv;
163 //! For each connection, keep a counter of how many requests are open
164 std::unordered_map<const evhttp_connection*, size_t> m_tracker GUARDED_BY(m_mutex);
165
166 void RemoveConnectionInternal(const decltype(m_tracker)::iterator it) EXCLUSIVE_LOCKS_REQUIRED(m_mutex)
167 {
168 m_tracker.erase(it);
169 if (m_tracker.empty()) m_cv.notify_all();
170 }
171 public:
172 //! Increase request counter for the associated connection by 1
173 void AddRequest(evhttp_request* req) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
174 {
175 const evhttp_connection* conn{Assert(evhttp_request_get_connection(Assert(req)))};
176 WITH_LOCK(m_mutex, ++m_tracker[conn]);
177 }
178 //! Decrease request counter for the associated connection by 1, remove connection if counter is 0
179 void RemoveRequest(evhttp_request* req) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
180 {
181 const evhttp_connection* conn{Assert(evhttp_request_get_connection(Assert(req)))};
182 LOCK(m_mutex);
183 auto it{m_tracker.find(conn)};
184 if (it != m_tracker.end() && it->second > 0) {
185 if (--(it->second) == 0) RemoveConnectionInternal(it);
186 }
187 }
188 //! Remove a connection entirely
189 void RemoveConnection(const evhttp_connection* conn) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
190 {
191 LOCK(m_mutex);
192 auto it{m_tracker.find(Assert(conn))};
193 if (it != m_tracker.end()) RemoveConnectionInternal(it);
194 }
195 size_t CountActiveConnections() const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
196 {
197 return WITH_LOCK(m_mutex, return m_tracker.size());
198 }
199 //! Wait until there are no more connections with active requests in the tracker
200 void WaitUntilEmpty() const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
201 {
202 WAIT_LOCK(m_mutex, lock);
203 m_cv.wait(lock, [this]() EXCLUSIVE_LOCKS_REQUIRED(m_mutex) { return m_tracker.empty(); });
204 }
205 };
206 //! Track active requests
207 static HTTPRequestTracker g_requests;
208
209 /** Check if a network address is allowed to access the HTTP server */
210 static bool ClientAllowed(const CNetAddr& netaddr)
211 {
212 if (!netaddr.IsValid())
213 return false;
214 for(const CSubNet& subnet : rpc_allow_subnets)
215 if (subnet.Match(netaddr))
216 return true;
217 return false;
218 }
219
220 /** Initialize ACL list for HTTP server */
221 static bool InitHTTPAllowList()
222 {
223 rpc_allow_subnets.clear();
224 rpc_allow_subnets.emplace_back(LookupHost("127.0.0.1", false).value(), 8); // always allow IPv4 local subnet
225 rpc_allow_subnets.emplace_back(LookupHost("::1", false).value()); // always allow IPv6 localhost
226 for (const std::string& strAllow : gArgs.GetArgs("-rpcallowip")) {
227 const CSubNet subnet{LookupSubNet(strAllow)};
228 if (!subnet.IsValid()) {
229 uiInterface.ThreadSafeMessageBox(
230 Untranslated(strprintf("Invalid -rpcallowip subnet specification: %s. Valid values are a single IP (e.g. 1.2.3.4), a network/netmask (e.g. 1.2.3.4/255.255.255.0), a network/CIDR (e.g. 1.2.3.4/24), all ipv4 (0.0.0.0/0), or all ipv6 (::/0). RFC4193 is allowed only if -cjdnsreachable=0.", strAllow)),
231 "", CClientUIInterface::MSG_ERROR);
232 return false;
233 }
234 rpc_allow_subnets.push_back(subnet);
235 }
236 std::string strAllowed;
237 for (const CSubNet& subnet : rpc_allow_subnets)
238 strAllowed += subnet.ToString() + " ";
239 LogDebug(BCLog::HTTP, "Allowing HTTP connections from: %s\n", strAllowed);
240 return true;
241 }
242
243 /** HTTP request method as string - use for logging only */
244 std::string RequestMethodString(HTTPRequest::RequestMethod m)
245 {
246 switch (m) {
247 case HTTPRequest::GET:
248 return "GET";
249 case HTTPRequest::POST:
250 return "POST";
251 case HTTPRequest::HEAD:
252 return "HEAD";
253 case HTTPRequest::PUT:
254 return "PUT";
255 case HTTPRequest::UNKNOWN:
256 return "unknown";
257 } // no default case, so the compiler can warn about missing cases
258 assert(false);
259 }
260
261 /** HTTP request callback */
262 static void http_request_cb(struct evhttp_request* req, void* arg)
263 {
264 evhttp_connection* conn{evhttp_request_get_connection(req)};
265 // Track active requests
266 {
267 g_requests.AddRequest(req);
268 evhttp_request_set_on_complete_cb(req, [](struct evhttp_request* req, void*) {
269 g_requests.RemoveRequest(req);
270 }, nullptr);
271 evhttp_connection_set_closecb(conn, [](evhttp_connection* conn, void* arg) {
272 g_requests.RemoveConnection(conn);
273 }, nullptr);
274 }
275
276 // Disable reading to work around a libevent bug, fixed in 2.1.9
277 // See https://github.com/libevent/libevent/commit/5ff8eb26371c4dc56f384b2de35bea2d87814779
278 // and https://github.com/limenka/limenka/pull/11593.
279 if (event_get_version_number() >= 0x02010600 && event_get_version_number() < 0x02010900) {
280 if (conn) {
281 bufferevent* bev = evhttp_connection_get_bufferevent(conn);
282 if (bev) {
283 bufferevent_disable(bev, EV_READ);
284 }
285 }
286 }
287 auto hreq{std::make_unique<HTTPRequest>(req, *static_cast<const util::SignalInterrupt*>(arg))};
288
289 // Early address-based allow check
290 if (!ClientAllowed(hreq->GetPeer())) {
291 LogDebug(BCLog::HTTP, "HTTP request from %s rejected: Client network is not allowed RPC access\n",
292 hreq->GetPeer().ToStringAddrPort());
293 hreq->WriteReply(HTTP_FORBIDDEN);
294 return;
295 }
296
297 // Early reject unknown HTTP methods
298 if (hreq->GetRequestMethod() == HTTPRequest::UNKNOWN) {
299 LogDebug(BCLog::HTTP, "HTTP request from %s rejected: Unknown HTTP request method\n",
300 hreq->GetPeer().ToStringAddrPort());
301 hreq->WriteReply(HTTP_BAD_METHOD);
302 return;
303 }
304
305 LogDebug(BCLog::HTTP, "Received a %s request for %s from %s\n",
306 RequestMethodString(hreq->GetRequestMethod()), SanitizeString(hreq->GetURI(), SAFE_CHARS_URI).substr(0, 100), hreq->GetPeer().ToStringAddrPort());
307
308 // Find registered handler for prefix
309 std::string strURI = hreq->GetURI();
310 std::string path;
311 LOCK(g_httppathhandlers_mutex);
312 std::vector<HTTPPathHandler>::const_iterator i = pathHandlers.begin();
313 std::vector<HTTPPathHandler>::const_iterator iend = pathHandlers.end();
314 for (; i != iend; ++i) {
315 bool match = false;
316 if (i->exactMatch)
317 match = (strURI == i->prefix);
318 else
319 match = strURI.starts_with(i->prefix);
320 if (match) {
321 path = strURI.substr(i->prefix.size());
322 break;
323 }
324 }
325
326 // Dispatch to worker thread
327 if (i != iend) {
328 std::unique_ptr<HTTPWorkItem> item(new HTTPWorkItem(std::move(hreq), path, i->handler));
329 assert(g_work_queue);
330 if (g_work_queue->Enqueue(item.get())) {
331 item.release(); /* if true, queue took ownership */
332 } else {
333 LogWarning("Request rejected because http work queue depth exceeded, it can be increased with the -rpcworkqueue= setting");
334 item->req->WriteReply(HTTP_SERVICE_UNAVAILABLE, "Work queue depth exceeded");
335 }
336 } else {
337 hreq->WriteReply(HTTP_NOT_FOUND);
338 }
339 }
340
341 /** Callback to reject HTTP requests after shutdown. */
342 static void http_reject_request_cb(struct evhttp_request* req, void*)
343 {
344 LogDebug(BCLog::HTTP, "Rejecting request while shutting down\n");
345 evhttp_send_error(req, HTTP_SERVUNAVAIL, nullptr);
346 }
347
348 /** Event dispatcher thread */
349 static void ThreadHTTP(struct event_base* base)
350 {
351 util::ThreadRename("http");
352 LogDebug(BCLog::HTTP, "Entering http event loop\n");
353 event_base_dispatch(base);
354 // Event loop will be interrupted by InterruptHTTPServer()
355 LogDebug(BCLog::HTTP, "Exited http event loop\n");
356 }
357
358 static struct evhttp_bound_socket *
359 my_bind_socket_with_handle(struct evhttp *http, const char *address, ev_uint16_t port, bool& ignorable_error)
360 {
361 evutil_socket_t fd;
362 struct evhttp_bound_socket *bound;
363 int serrno;
364
365 struct evutil_addrinfo *aitop = nullptr;
366
367 if (address == nullptr && port == 0) {
368 fd = socket(AF_INET, SOCK_STREAM, 0);
369 if (fd == -1) {
370 LogPrintf("libevent: socket: %s\n", evutil_socket_error_to_string(evutil_socket_geterror(-1)));
371 return nullptr;
372 }
373 } else {
374 struct evutil_addrinfo hints = {};
375 int ai_result;
376
377 hints.ai_family = AF_UNSPEC;
378 hints.ai_socktype = SOCK_STREAM;
379 hints.ai_flags = EVUTIL_AI_PASSIVE|EVUTIL_AI_ADDRCONFIG;
380 const std::string strport = strprintf("%d", port);
381 ai_result = evutil_getaddrinfo(address, strport.c_str(), &hints, &aitop);
382 if (ai_result || !aitop) {
383 switch (ai_result) {
384 case 0:
385 break;
386 case EVUTIL_EAI_SYSTEM:
387 LogPrintf("libevent: getaddrinfo\n");
388 break;
389 case EVUTIL_EAI_NODATA:
390 LogPrintf("evutil_getaddrinfo doesn't support IPv6; cannot bind %s:%d\n", address, port);
391 [[fallthrough]];
392 case EVUTIL_EAI_ADDRFAMILY:
393 case EVUTIL_EAI_FAMILY:
394 case EVUTIL_EAI_SOCKTYPE:
395 ignorable_error = true;
396 break;
397 default:
398 LogPrintf("libevent: getaddrinfo: %s\n", evutil_gai_strerror(ai_result));
399 }
400 return nullptr;
401 }
402
403 fd = socket(aitop->ai_family, SOCK_STREAM, 0);
404 if (fd == -1) {
405 evutil_freeaddrinfo(aitop);
406 return nullptr;
407 }
408 evutil_make_listen_socket_reuseable(fd);
409 }
410
411 const int on = 1;
412 setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (sockopt_arg_type)&on, sizeof(on));
413
414 bool listen_failed = false;
415 if (evutil_make_socket_nonblocking(fd) < 0 ||
416 evutil_make_socket_closeonexec(fd) < 0 ||
417 (aitop && bind(fd, aitop->ai_addr, aitop->ai_addrlen) == -1) ||
418 (listen_failed = (listen(fd, 128) == -1))
419 ) {
420 serrno = EVUTIL_SOCKET_ERROR();
421 if (listen_failed) LogPrintf("libevent: %s: listen\n", __func__);
422 evutil_closesocket(fd);
423 if (aitop) evutil_freeaddrinfo(aitop);
424 EVUTIL_SET_SOCKET_ERROR(serrno);
425 return nullptr;
426 }
427
428 if (aitop) evutil_freeaddrinfo(aitop);
429
430 bound = evhttp_accept_socket_with_handle(http, fd);
431 if (bound == nullptr) {
432 evutil_closesocket(fd);
433 return nullptr;
434 }
435
436 LogDebug(BCLog::LIBEVENT, "libevent: Bound to port %d - Awaiting connections ... \n", port);
437 return bound;
438 }
439
440 /** Bind HTTP server to specified addresses */
441 static bool HTTPBindAddresses(struct evhttp* http)
442 {
443 uint16_t http_port{static_cast<uint16_t>(gArgs.GetIntArg("-rpcport", BaseParams().RPCPort()))};
444 std::vector<std::pair<std::string, uint16_t>> endpoints;
445 bool is_default = false;
446
447 // Determine what addresses to bind to
448 // To prevent misconfiguration and accidental exposure of the RPC
449 // interface, require -rpcallowip and -rpcbind to both be specified
450 // together. If either is missing, ignore both values, bind to localhost
451 // instead, and log warnings.
452 if (gArgs.GetArgs("-rpcallowip").empty() || gArgs.GetArgs("-rpcbind").empty()) { // Default to loopback if not allowing external IPs
453 endpoints.emplace_back("::1", http_port);
454 endpoints.emplace_back("127.0.0.1", http_port);
455 is_default = true;
456 if (!gArgs.GetArgs("-rpcallowip").empty()) {
457 LogWarning("Option -rpcallowip was specified without -rpcbind; this doesn't usually make sense");
458 }
459 if (!gArgs.GetArgs("-rpcbind").empty()) {
460 InitWarning(_("Option -rpcbind was ignored because -rpcallowip was not specified, refusing to allow everyone to connect\n"));
461 }
462 } else { // Specific bind addresses
463 for (const std::string& strRPCBind : gArgs.GetArgs("-rpcbind")) {
464 uint16_t port{http_port};
465 std::string host;
466 if (!SplitHostPort(strRPCBind, port, host)) {
467 LogError("%s\n", InvalidPortErrMsg("-rpcbind", strRPCBind).original);
468 return false;
469 }
470 endpoints.emplace_back(host, port);
471 }
472 }
473
474 // Bind addresses
475 int num_fail = 0;
476 for (std::vector<std::pair<std::string, uint16_t> >::iterator i = endpoints.begin(); i != endpoints.end(); ++i) {
477 LogPrintf("Binding RPC on address %s port %i\n", i->first, i->second);
478 bool ignorable_error = false;
479 evhttp_bound_socket *bind_handle = my_bind_socket_with_handle(http, i->first.empty() ? nullptr : i->first.c_str(), i->second, ignorable_error);
480 if (bind_handle) {
481 const std::optional<CNetAddr> addr{LookupHost(i->first, false)};
482 if (i->first.empty() || (addr.has_value() && addr->IsBindAny())) {
483 LogWarning("The RPC server is not safe to expose to untrusted networks such as the public internet");
484 }
485 // Set the no-delay option (disable Nagle's algorithm) on the TCP socket.
486 evutil_socket_t fd = evhttp_bound_socket_get_fd(bind_handle);
487 int one = 1;
488 if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (sockopt_arg_type)&one, sizeof(one)) == SOCKET_ERROR) {
489 LogInfo("WARNING: Unable to set TCP_NODELAY on RPC server socket, continuing anyway\n");
490 }
491 boundSockets.push_back(bind_handle);
492 } else {
493 int err = EVUTIL_SOCKET_ERROR();
494 if (!is_default || (err != EADDRNOTAVAIL && err != ENOENT && err != EOPNOTSUPP && !ignorable_error)) {
495 LogWarning("Binding RPC on address %s port %i failed (Error: %s).", i->first, i->second, NetworkErrorString(err));
496 num_fail += 1;
497 } else {
498 // Don't count failure if binding was not explicitly configured
499 // (default settings) and the address is not available.
500 // (for example: Travis without IPv6 localhost will return ENOENT)
501 LogPrintf("Binding RPC on address %s port %i failed, error ignored because interface was unavailable.\n", i->first, i->second);
502 }
503 }
504 }
505 if (num_fail != 0) {
506 // In case of an error, clean up listening sockets that succeeded to
507 // avoid leak
508 for (evhttp_bound_socket *socket : boundSockets) {
509 evhttp_del_accept_socket(http, socket);
510 }
511 boundSockets.clear();
512 }
513 return num_fail == 0;
514 }
515
516 /** Simple wrapper to set thread name and run work queue */
517 static void HTTPWorkQueueRun(WorkQueue<HTTPClosure>* queue, int worker_num)
518 {
519 util::ThreadRename(strprintf("httpworker.%i", worker_num));
520 queue->Run();
521 }
522
523 /** libevent event log callback */
524 static void libevent_log_cb(int severity, const char *msg)
525 {
526 BCLog::Level level;
527 switch (severity) {
528 case EVENT_LOG_DEBUG:
529 level = BCLog::Level::Debug;
530 break;
531 case EVENT_LOG_MSG:
532 level = BCLog::Level::Info;
533 break;
534 case EVENT_LOG_WARN:
535 level = BCLog::Level::Warning;
536 break;
537 default: // EVENT_LOG_ERR and others are mapped to error
538 level = BCLog::Level::Error;
539 break;
540 }
541 LogPrintLevel(BCLog::LIBEVENT, level, "%s\n", msg);
542 }
543
544 bool InitHTTPServer(const util::SignalInterrupt& interrupt)
545 {
546 if (!InitHTTPAllowList())
547 return false;
548
549 // Redirect libevent's logging to our own log
550 event_set_log_callback(&libevent_log_cb);
551 // Update libevent's log handling.
552 UpdateHTTPServerLogging(LogInstance().WillLogCategory(BCLog::LIBEVENT));
553
554 #ifdef WIN32
555 evthread_use_windows_threads();
556 #else
557 evthread_use_pthreads();
558 #endif
559
560 raii_event_base base_ctr = obtain_event_base();
561
562 /* Create a new evhttp object to handle requests. */
563 raii_evhttp http_ctr = obtain_evhttp(base_ctr.get());
564 struct evhttp* http = http_ctr.get();
565 if (!http) {
566 LogError("Couldn't create evhttp. Exiting.");
567 return false;
568 }
569
570 evhttp_set_timeout(http, gArgs.GetIntArg("-rpcservertimeout", DEFAULT_HTTP_SERVER_TIMEOUT));
571 evhttp_set_max_headers_size(http, MAX_HEADERS_SIZE);
572 evhttp_set_max_body_size(http, MAX_SIZE);
573 evhttp_set_gencb(http, http_request_cb, (void*)&interrupt);
574
575 if (!HTTPBindAddresses(http)) {
576 LogError("Unable to bind all endpoints for RPC server");
577 return false;
578 }
579
580 LogDebug(BCLog::HTTP, "Initialized HTTP server\n");
581 int workQueueDepth = std::max((long)gArgs.GetIntArg("-rpcworkqueue", DEFAULT_HTTP_WORKQUEUE), 1L);
582 LogDebug(BCLog::HTTP, "creating work queue of depth %d\n", workQueueDepth);
583
584 g_work_queue = std::make_unique<WorkQueue<HTTPClosure>>(workQueueDepth);
585 // transfer ownership to eventBase/HTTP via .release()
586 eventBase = base_ctr.release();
587 eventHTTP = http_ctr.release();
588 return true;
589 }
590
591 void UpdateHTTPServerLogging(bool enable) {
592 if (enable) {
593 event_enable_debug_logging(EVENT_DBG_ALL);
594 } else {
595 event_enable_debug_logging(EVENT_DBG_NONE);
596 }
597 }
598
599 static std::thread g_thread_http;
600 static std::vector<std::thread> g_thread_http_workers;
601
602 void StartHTTPServer()
603 {
604 int rpcThreads = std::max((long)gArgs.GetIntArg("-rpcthreads", DEFAULT_HTTP_THREADS), 1L);
605 LogInfo("Starting HTTP server with %d worker threads\n", rpcThreads);
606 g_thread_http = std::thread(ThreadHTTP, eventBase);
607
608 for (int i = 0; i < rpcThreads; i++) {
609 g_thread_http_workers.emplace_back(HTTPWorkQueueRun, g_work_queue.get(), i);
610 }
611 }
612
613 void InterruptHTTPServer()
614 {
615 LogDebug(BCLog::HTTP, "Interrupting HTTP server\n");
616 if (eventHTTP) {
617 // Reject requests on current connections
618 evhttp_set_gencb(eventHTTP, http_reject_request_cb, nullptr);
619 }
620 if (g_work_queue) {
621 g_work_queue->Interrupt();
622 }
623 }
624
625 void StopHTTPServer()
626 {
627 LogDebug(BCLog::HTTP, "Stopping HTTP server\n");
628 if (g_work_queue) {
629 LogDebug(BCLog::HTTP, "Waiting for HTTP worker threads to exit\n");
630 for (auto& thread : g_thread_http_workers) {
631 thread.join();
632 }
633 g_thread_http_workers.clear();
634 }
635 // Unlisten sockets, these are what make the event loop running, which means
636 // that after this and all connections are closed the event loop will quit.
637 for (evhttp_bound_socket *socket : boundSockets) {
638 evhttp_del_accept_socket(eventHTTP, socket);
639 }
640 boundSockets.clear();
641 {
642 if (const auto n_connections{g_requests.CountActiveConnections()}; n_connections != 0) {
643 LogDebug(BCLog::HTTP, "Waiting for %d connections to stop HTTP server\n", n_connections);
644 }
645 g_requests.WaitUntilEmpty();
646 }
647 if (eventHTTP) {
648 // Schedule a callback to call evhttp_free in the event base thread, so
649 // that evhttp_free does not need to be called again after the handling
650 // of unfinished request connections that follows.
651 event_base_once(eventBase, -1, EV_TIMEOUT, [](evutil_socket_t, short, void*) {
652 evhttp_free(eventHTTP);
653 eventHTTP = nullptr;
654 }, nullptr, nullptr);
655 }
656 if (eventBase) {
657 LogDebug(BCLog::HTTP, "Waiting for HTTP event thread to exit\n");
658 if (g_thread_http.joinable()) g_thread_http.join();
659 event_base_free(eventBase);
660 eventBase = nullptr;
661 }
662 g_work_queue.reset();
663 LogDebug(BCLog::HTTP, "Stopped HTTP server\n");
664 }
665
666 struct event_base* EventBase()
667 {
668 return eventBase;
669 }
670
671 static void httpevent_callback_fn(evutil_socket_t, short, void* data)
672 {
673 // Static handler: simply call inner handler
674 HTTPEvent *self = static_cast<HTTPEvent*>(data);
675 self->handler();
676 if (self->deleteWhenTriggered)
677 delete self;
678 }
679
680 HTTPEvent::HTTPEvent(struct event_base* base, bool _deleteWhenTriggered, const std::function<void()>& _handler):
681 deleteWhenTriggered(_deleteWhenTriggered), handler(_handler)
682 {
683 ev = event_new(base, -1, 0, httpevent_callback_fn, this);
684 assert(ev);
685 }
686 HTTPEvent::~HTTPEvent()
687 {
688 event_free(ev);
689 }
690 void HTTPEvent::trigger(struct timeval* tv)
691 {
692 if (tv == nullptr)
693 event_active(ev, 0, 0); // immediately trigger event in main thread
694 else
695 evtimer_add(ev, tv); // trigger after timeval passed
696 }
697 HTTPRequest::HTTPRequest(struct evhttp_request* _req, const util::SignalInterrupt& interrupt, bool _replySent)
698 : req(_req), m_interrupt(interrupt), replySent(_replySent)
699 {
700 }
701
702 HTTPRequest::~HTTPRequest()
703 {
704 if (!replySent) {
705 // Keep track of whether reply was sent to avoid request leaks
706 LogWarning("Unhandled HTTP request");
707 WriteReply(HTTP_INTERNAL_SERVER_ERROR, "Unhandled request");
708 }
709 // evhttpd cleans up the request, as long as a reply was sent.
710 }
711
712 std::pair<bool, std::string> HTTPRequest::GetHeader(const std::string& hdr) const
713 {
714 const struct evkeyvalq* headers = evhttp_request_get_input_headers(req);
715 assert(headers);
716 const char* val = evhttp_find_header(headers, hdr.c_str());
717 if (val)
718 return std::make_pair(true, val);
719 else
720 return std::make_pair(false, "");
721 }
722
723 std::string HTTPRequest::ReadBody()
724 {
725 struct evbuffer* buf = evhttp_request_get_input_buffer(req);
726 if (!buf)
727 return "";
728 size_t size = evbuffer_get_length(buf);
729 /** Trivial implementation: if this is ever a performance bottleneck,
730 * internal copying can be avoided in multi-segment buffers by using
731 * evbuffer_peek and an awkward loop. Though in that case, it'd be even
732 * better to not copy into an intermediate string but use a stream
733 * abstraction to consume the evbuffer on the fly in the parsing algorithm.
734 */
735 const char* data = (const char*)evbuffer_pullup(buf, size);
736 if (!data) // returns nullptr in case of empty buffer
737 return "";
738 std::string rv(data, size);
739 evbuffer_drain(buf, size);
740 return rv;
741 }
742
743 void HTTPRequest::WriteHeader(const std::string& hdr, const std::string& value)
744 {
745 struct evkeyvalq* headers = evhttp_request_get_output_headers(req);
746 assert(headers);
747 evhttp_add_header(headers, hdr.c_str(), value.c_str());
748 }
749
750 /** Closure sent to main thread to request a reply to be sent to
751 * a HTTP request.
752 * Replies must be sent in the main loop in the main http thread,
753 * this cannot be done from worker threads.
754 */
755 void HTTPRequest::WriteReply(int nStatus, std::span<const std::byte> reply)
756 {
757 assert(!replySent && req);
758 if (m_interrupt) {
759 WriteHeader("Connection", "close");
760 }
761 // Send event to main http thread to send reply message
762 struct evbuffer* evb = evhttp_request_get_output_buffer(req);
763 assert(evb);
764 evbuffer_add(evb, reply.data(), reply.size());
765 auto req_copy = req;
766 HTTPEvent* ev = new HTTPEvent(eventBase, true, [req_copy, nStatus]{
767 evhttp_send_reply(req_copy, nStatus, nullptr, nullptr);
768 // Re-enable reading from the socket. This is the second part of the libevent
769 // workaround above.
770 if (event_get_version_number() >= 0x02010600 && event_get_version_number() < 0x02010900) {
771 evhttp_connection* conn = evhttp_request_get_connection(req_copy);
772 if (conn) {
773 bufferevent* bev = evhttp_connection_get_bufferevent(conn);
774 if (bev) {
775 bufferevent_enable(bev, EV_READ | EV_WRITE);
776 }
777 }
778 }
779 });
780 ev->trigger(nullptr);
781 replySent = true;
782 req = nullptr; // transferred back to main thread
783 }
784
785 CService HTTPRequest::GetPeer() const
786 {
787 evhttp_connection* con = evhttp_request_get_connection(req);
788 CService peer;
789 if (con) {
790 // evhttp retains ownership over returned address string
791 const char* address = "";
792 uint16_t port = 0;
793
794 #ifdef HAVE_EVHTTP_CONNECTION_GET_PEER_CONST_CHAR
795 evhttp_connection_get_peer(con, &address, &port);
796 #else
797 evhttp_connection_get_peer(con, (char**)&address, &port);
798 #endif // HAVE_EVHTTP_CONNECTION_GET_PEER_CONST_CHAR
799
800 peer = MaybeFlipIPv6toCJDNS(LookupNumeric(address, port));
801 }
802 return peer;
803 }
804
805 std::string HTTPRequest::GetURI() const
806 {
807 return evhttp_request_get_uri(req);
808 }
809
810 HTTPRequest::RequestMethod HTTPRequest::GetRequestMethod() const
811 {
812 switch (evhttp_request_get_command(req)) {
813 case EVHTTP_REQ_GET:
814 return GET;
815 case EVHTTP_REQ_POST:
816 return POST;
817 case EVHTTP_REQ_HEAD:
818 return HEAD;
819 case EVHTTP_REQ_PUT:
820 return PUT;
821 default:
822 return UNKNOWN;
823 }
824 }
825
826 std::optional<std::string> HTTPRequest::GetQueryParameter(const std::string& key) const
827 {
828 const char* uri{evhttp_request_get_uri(req)};
829
830 return GetQueryParameterFromUri(uri, key);
831 }
832
833 std::optional<std::string> GetQueryParameterFromUri(const char* uri, const std::string& key)
834 {
835 evhttp_uri* uri_parsed{evhttp_uri_parse(uri)};
836 if (!uri_parsed) {
837 throw std::runtime_error("URI parsing failed, it likely contained RFC 3986 invalid characters");
838 }
839 const char* query{evhttp_uri_get_query(uri_parsed)};
840 std::optional<std::string> result;
841
842 if (query) {
843 // Parse the query string into a key-value queue and iterate over it
844 struct evkeyvalq params_q;
845 evhttp_parse_query_str(query, ¶ms_q);
846
847 for (struct evkeyval* param{params_q.tqh_first}; param != nullptr; param = param->next.tqe_next) {
848 if (param->key == key) {
849 result = param->value;
850 break;
851 }
852 }
853 evhttp_clear_headers(¶ms_q);
854 }
855 evhttp_uri_free(uri_parsed);
856
857 return result;
858 }
859
860 void RegisterHTTPHandler(const std::string &prefix, bool exactMatch, const HTTPRequestHandler &handler)
861 {
862 LogDebug(BCLog::HTTP, "Registering HTTP handler for %s (exactmatch %d)\n", prefix, exactMatch);
863 LOCK(g_httppathhandlers_mutex);
864 pathHandlers.emplace_back(prefix, exactMatch, handler);
865 }
866
867 void UnregisterHTTPHandler(const std::string &prefix, bool exactMatch)
868 {
869 LOCK(g_httppathhandlers_mutex);
870 std::vector<HTTPPathHandler>::iterator i = pathHandlers.begin();
871 std::vector<HTTPPathHandler>::iterator iend = pathHandlers.end();
872 for (; i != iend; ++i)
873 if (i->prefix == prefix && i->exactMatch == exactMatch)
874 break;
875 if (i != iend)
876 {
877 LogDebug(BCLog::HTTP, "Unregistering HTTP handler for %s (exactmatch %d)\n", prefix, exactMatch);
878 pathHandlers.erase(i);
879 }
880 }
881