Line data Source code
1 : #include "listener.hpp"
2 : #include "async_event.hpp"
3 : #include "detail/listener_access.hpp"
4 : #include "io_context_of.hpp"
5 : #include "with_timeout.hpp"
6 : #include "worker_thread.hpp"
7 : #include "sockets/ssl_socket.hpp"
8 : #include "sockets/tcp_socket.hpp"
9 : #include "sockets/unix_socket.hpp"
10 : #include "ssl/certificate_manager.hpp"
11 : #include "ssl/client_verify.hpp"
12 : #include "../util/logger.hpp"
13 : #include <algorithm>
14 : #include <atomic>
15 : #include <cerrno>
16 : #include <condition_variable>
17 : #include <exception>
18 : #include <filesystem>
19 : #include <mutex>
20 : #include <optional>
21 : #include <set>
22 : #include <system_error>
23 : #include <thread>
24 : #include <type_traits>
25 : #include <unordered_map>
26 : #include <utility>
27 : #include <boost/asio/ip/tcp.hpp>
28 : #include <boost/asio/local/stream_protocol.hpp>
29 : #include <boost/asio/post.hpp>
30 : #include <boost/asio/steady_timer.hpp>
31 : #include <boost/asio/strand.hpp>
32 : #include <boost/asio/this_coro.hpp>
33 : #include <fcntl.h>
34 : #include <sys/socket.h>
35 : #include <unistd.h>
36 :
37 : namespace thinger::asio {
38 :
39 : namespace {
40 :
41 : using tcp = boost::asio::ip::tcp;
42 : using local = boost::asio::local::stream_protocol;
43 : using steady = std::chrono::steady_clock;
44 :
45 : // Time between attempts to bind (see listener_options::max_listening_attempts)
46 : constexpr std::chrono::seconds LISTEN_RETRY_DELAY{5};
47 : // Time before accepting again after an error accepting (e.g. out of descriptors)
48 : constexpr std::chrono::seconds ACCEPT_RETRY_DELAY{1};
49 :
50 : // An accepted descriptor on its way to the executor serving it, closed if it never gets there
51 : class accepted_descriptor {
52 : public:
53 17256 : explicit accepted_descriptor(int descriptor) : descriptor_(descriptor) {}
54 103506 : accepted_descriptor(accepted_descriptor&& other) noexcept : descriptor_(std::exchange(other.descriptor_, -1)) {}
55 : accepted_descriptor& operator=(accepted_descriptor&&) = delete;
56 120762 : ~accepted_descriptor() { if (descriptor_ >= 0) ::close(descriptor_); }
57 :
58 34506 : int get() const { return descriptor_; }
59 17250 : int release() { return std::exchange(descriptor_, -1); }
60 :
61 : private:
62 : int descriptor_;
63 : };
64 :
65 : // A socket of the listener is not inherited by the programs a handler runs (fork and exec): asio
66 : // opens it without SOCK_CLOEXEC. The acceptor thread on Linux needs none (see accept_pending())
67 43761 : void close_on_exec(int descriptor) {
68 43761 : ::fcntl(descriptor, F_SETFD, FD_CLOEXEC);
69 43761 : }
70 :
71 : // Sets up and serves an accepted connection. Nothing doing it may throw out of an accept handler
72 : // or a worker: it would end the process. What is thrown is logged, and only drops the connection
73 : // (its socket is destroyed, and closed, while unwinding)
74 : template<typename Function>
75 80817 : void serve_without_throwing(Function&& serve) noexcept {
76 : try {
77 80817 : serve();
78 24 : } catch (const std::exception& e) {
79 12 : LOG_ERROR("cannot serve an accepted connection: {}", e.what());
80 0 : } catch (...) {
81 0 : LOG_ERROR("cannot serve an accepted connection");
82 : }
83 80817 : }
84 :
85 : // The thread a listener accepts on (see detail::listener_access): stopped, and joined, before
86 : // it is destroyed
87 : struct thread_stopper {
88 15235 : void operator()(worker_thread* thread) const {
89 15235 : thread->stop();
90 15235 : delete thread;
91 15235 : }
92 : };
93 : using acceptor_thread_ptr = std::unique_ptr<worker_thread, thread_stopper>;
94 :
95 : // The TLS of the connections of a listener, taken by open(): each one is created with this
96 : // context (its handshake may switch to another certificate of the manager, see
97 : // certificate_manager::sni_callback()) and asks for client certificates with this policy
98 : struct tls_setup {
99 : std::shared_ptr<boost::asio::ssl::context> context;
100 : std::shared_ptr<const client_verify_policy> client_verify;
101 : };
102 :
103 : // The TLS handshakes in progress on one io_context serving a listener with a bound (see
104 : // listener_options::max_pending_handshakes). Used only on its strand, where the handshakes it
105 : // bounds run: the connections waiting for a slot wait for `slot_freed`, set whenever a slot is
106 : // freed (all of them wake and check again: the order is not first come, first served)
107 : struct handshake_gate {
108 36 : explicit handshake_gate(boost::asio::io_context& io_context)
109 36 : : strand(boost::asio::make_strand(io_context)), slot_freed(strand) {}
110 :
111 : boost::asio::any_io_executor strand;
112 : async_event slot_freed;
113 : size_t in_progress = 0;
114 : };
115 :
116 : // A slot of a gate, taken while a handshake runs: freed (waking the connections waiting for one)
117 : // however the handshake ends
118 : class handshake_slot {
119 : public:
120 114 : explicit handshake_slot(std::shared_ptr<handshake_gate> gate) : gate_(std::move(gate)) { ++gate_->in_progress; }
121 114 : ~handshake_slot() {
122 114 : --gate_->in_progress;
123 114 : gate_->slot_freed.set();
124 114 : }
125 : handshake_slot(const handshake_slot&) = delete;
126 : handshake_slot& operator=(const handshake_slot&) = delete;
127 :
128 : private:
129 : std::shared_ptr<handshake_gate> gate_;
130 : };
131 :
132 : // A connection accepted whose handshake has not ended (waiting for a slot included): stop()
133 : // closes it on the executor it handshakes on
134 : struct pending_handshake {
135 : std::weak_ptr<tcp_socket> sock;
136 : boost::asio::any_io_executor executor;
137 : std::shared_ptr<handshake_gate> gate; // none without a bound
138 : };
139 :
140 : // The handshake of a TLS connection, closed if it does not end by `deadline` (none: no limit)
141 8259 : awaitable<boost::system::error_code> handshake_until(std::shared_ptr<tcp_socket> sock,
142 : std::optional<std::chrono::steady_clock::time_point> deadline) {
143 : if (!deadline) co_return co_await sock->handshake();
144 : if (*deadline <= std::chrono::steady_clock::now()) co_return boost::asio::error::timed_out;
145 : // On the executor of the handshake. The handler does not own the socket
146 : auto timer = std::make_shared<boost::asio::steady_timer>(co_await boost::asio::this_coro::executor, *deadline);
147 : auto expired = std::make_shared<std::atomic<bool>>(false);
148 8134 : timer->async_wait([weak = std::weak_ptr<tcp_socket>(sock), expired](const boost::system::error_code& ec) {
149 8134 : if (ec) return;
150 3042 : *expired = true;
151 3042 : if (auto pending = weak.lock()) pending->close();
152 : });
153 : auto ec = co_await sock->handshake();
154 : timer->cancel();
155 : co_return *expired ? boost::system::error_code(boost::asio::error::timed_out) : ec;
156 16518 : }
157 :
158 : // What a TLS connection throws is logged, as when a connection is served right away (see
159 : // serve_without_throwing())
160 8185 : void log_serve_failure(const std::exception_ptr& error) {
161 8185 : if (!error) return;
162 : try {
163 0 : std::rethrow_exception(error);
164 0 : } catch (const std::exception& e) {
165 0 : LOG_ERROR("cannot serve a TLS connection: {}", e.what());
166 0 : } catch (...) {
167 0 : LOG_ERROR("cannot serve a TLS connection");
168 0 : }
169 : }
170 :
171 : } // namespace
172 :
173 : // What a listener shares with the handlers accepting and serving its connections: they hold
174 : // it, never the listener, which may be gone by the time they run.
175 : //
176 : // Its mutex guards the short transitions only (stopping, the deliveries in progress, the
177 : // handshakes pending, the gates, the acceptor's use by an accept handler), and is never held
178 : // while the handler of the connections runs, nor while a handshake or a wait for a slot does:
179 : // the handler may stop the listener, or destroy it. Nor while the executor providers run: they
180 : // are the application's, and may block or call the listener (e.g. url()). Every connection reaches the handler
181 : // through deliver(), which records the thread delivering it unless the listener is stopping;
182 : // stop() sets stopping, closes the handshakes pending and waits until no delivery is recorded
183 : // (its own excepted), so that once it returns no handler starts and none runs on another
184 : // thread.
185 : struct listener::state : std::enable_shared_from_this<state> {
186 25307 : state(listener_options listener_options, executor_provider accept_executors, executor_provider serve_executors)
187 25307 : : certificates(std::exchange(listener_options.tls, nullptr))
188 25307 : , manager(certificates)
189 25307 : , secure(certificates != nullptr)
190 25307 : , options(std::move(listener_options))
191 25307 : , accept_on(std::move(accept_executors))
192 25307 : , serve_on(std::move(serve_executors))
193 25307 : , allowed_remotes(options.allowed_remotes.begin(), options.allowed_remotes.end())
194 50614 : , forbidden_remotes(options.forbidden_remotes.begin(), options.forbidden_remotes.end()) {}
195 :
196 : // The manager of the options (taken out of them), held until stop(), whatever open()
197 : // returns: the connections reach it through their contexts (see
198 : // certificate_manager::sni_callback()), so their host names keep their certificates and
199 : // client certificate modes while the listener runs, even if the application keeps none
200 : std::shared_ptr<certificate_manager> certificates;
201 : const std::weak_ptr<certificate_manager> manager;
202 : const bool secure;
203 : // The options, without the manager
204 : const listener_options options;
205 : const executor_provider accept_on;
206 : const executor_provider serve_on;
207 : const std::set<std::string> allowed_remotes;
208 : const std::set<std::string> forbidden_remotes;
209 : // Accept on a thread of its own (see detail::listener_access), set before open()
210 : bool own_acceptor_thread = false;
211 :
212 : mutable std::mutex mutex;
213 : // Notified when a delivery ends and when the listener stops
214 : std::condition_variable changed;
215 : // Written with the mutex held; read without it by the loops that only need to end
216 : std::atomic<bool> stopping{false};
217 : std::atomic<uint16_t> port{0};
218 : bool opened = false;
219 : bool started = false;
220 : std::string url;
221 : // The threads delivering a connection to the handler, one entry per delivery
222 : std::vector<std::thread::id> delivering;
223 : // The TLS connections from their accept to the end of their handshake; taken by stop()
224 : std::unordered_map<const tcp_socket*, pending_handshake> pending;
225 : // The gate of each io_context serving TLS connections, with a bound: created as they are
226 : // first used, kept until stop()
227 : std::unordered_map<boost::asio::io_context*, std::shared_ptr<handshake_gate>> gates;
228 : // Set by open(), with the mutex held, before anything accepts: no context without TLS
229 : tls_setup tls;
230 : // Whether it holds its session cache on the manager of `tls` (see open()): released once
231 : bool holds_session_cache = false;
232 :
233 : // Set by open(), with the mutex held, before anything accepts; taken by stop(). The thread
234 : // is declared before the acceptors, which use its io_context
235 : acceptor_thread_ptr acceptor_thread;
236 : std::unique_ptr<tcp::acceptor> tcp_acceptor;
237 : std::unique_ptr<local::acceptor> unix_acceptor;
238 : tcp tcp_protocol = tcp::v4();
239 : // Set by start(), before anything accepts
240 : sync_callback<void(std::shared_ptr<socket>)> on_connection;
241 :
242 52168 : bool is_tcp() const { return std::holds_alternative<tcp_endpoint>(options.endpoint); }
243 :
244 : template<typename Protocol>
245 123492 : std::unique_ptr<typename Protocol::acceptor>& acceptor_of() {
246 56704 : if constexpr (std::is_same_v<Protocol, tcp>) return tcp_acceptor;
247 66788 : else return unix_acceptor;
248 : }
249 :
250 : // The port of the options: 0 for Unix sockets
251 40553 : uint16_t configured_port() const {
252 40553 : const auto* where = std::get_if<tcp_endpoint>(&options.endpoint);
253 40553 : return where ? where->port : 0;
254 : }
255 :
256 : // "http://host:port", "https://host:port" or "unix:/path"
257 87689 : std::string url_for(uint16_t bound_port) const {
258 87689 : if (const auto* unix = std::get_if<unix_endpoint>(&options.endpoint)) return "unix:" + unix->path;
259 44771 : const auto& host = std::get<tcp_endpoint>(options.endpoint).host;
260 44771 : auto shown = host.find(':') == std::string::npos ? host : "[" + host + "]";
261 44771 : return (secure ? "https://" : "http://") + shown + ":" + std::to_string(bound_port);
262 44771 : }
263 :
264 : // --- open() ---
265 :
266 25310 : boost::system::error_code open() {
267 25310 : std::shared_ptr<certificate_manager> tls;
268 : {
269 25310 : std::scoped_lock lock(mutex);
270 25310 : if (opened || stopping) return boost::asio::error::already_open;
271 25303 : tls = certificates;
272 25310 : }
273 25303 : return open_with(tls);
274 25310 : }
275 :
276 25303 : boost::system::error_code open_with(const std::shared_ptr<certificate_manager>& tls) {
277 25303 : tls_setup connections_tls;
278 25303 : if (auto ec = prepare_tls(tls.get(), connections_tls)) return ec;
279 25279 : if (auto ec = claim_session_cache(tls.get())) return ec;
280 25258 : auto ec = open_claimed(std::move(connections_tls));
281 : // Nothing bound: the setting is left to other listeners
282 25258 : if (ec) release_session_cache();
283 25258 : return ec;
284 25303 : }
285 :
286 25258 : boost::system::error_code open_claimed(tls_setup connections_tls) {
287 25258 : acceptor_thread_ptr thread;
288 25258 : if (own_acceptor_thread) {
289 15235 : thread.reset(new worker_thread(url_for(configured_port()) + " acceptor"));
290 15235 : thread->start();
291 : }
292 65751 : auto executor = thread ? boost::asio::any_io_executor(thread->get_io_context().get_executor())
293 40493 : : (accept_on ? accept_on() : boost::asio::any_io_executor());
294 25258 : if (auto ec = check_executors(executor)) return ec;
295 77892 : return is_tcp() ? open_on<tcp>(executor, std::move(thread), std::move(connections_tls))
296 77892 : : open_on<local>(executor, std::move(thread), std::move(connections_tls));
297 25258 : }
298 :
299 : // The TLS of the connections, if any: the default certificate of the manager as it is now,
300 : // and the client certificate policy, whose trusted CAs must be readable
301 25303 : boost::system::error_code prepare_tls(certificate_manager* tls, tls_setup& setup) const {
302 25303 : if (!tls) {
303 21472 : if (options.client_verify.mode != client_verify::none || !options.client_verify.trusted_cas.empty()) {
304 0 : LOG_WARNING("{} has no TLS: its client certificate options are ignored", url_for(configured_port()));
305 : }
306 21472 : return {};
307 : }
308 3831 : if (!is_tcp()) {
309 0 : LOG_ERROR("cannot listen on {}: TLS is served on TCP endpoints only", url_for(configured_port()));
310 0 : return boost::asio::error::invalid_argument;
311 : }
312 3831 : setup.client_verify = client_verify_policy::create(options.client_verify);
313 3831 : if (!setup.client_verify) {
314 6 : LOG_ERROR("cannot listen on {}: the trusted CAs of the client certificates cannot be read",
315 : url_for(configured_port()));
316 6 : return boost::asio::error::invalid_argument;
317 : }
318 3825 : setup.context = tls->get_default_certificate();
319 3825 : if (!setup.context) {
320 18 : LOG_ERROR("cannot listen on {}: its certificate_manager has no default certificate", url_for(configured_port()));
321 18 : return boost::asio::error::invalid_argument;
322 : }
323 3807 : return {};
324 : }
325 :
326 : // Apply the session cache of the options to the manager, before binding: refused if other
327 : // listeners of the same manager hold another one. Plain listeners have none
328 25279 : boost::system::error_code claim_session_cache(certificate_manager* tls) {
329 25279 : if (!tls) return {};
330 3807 : if (auto ec = tls->claim_session_cache(options.session_cache)) {
331 21 : LOG_ERROR("cannot listen on {}: invalid TLS session cache", url_for(configured_port()));
332 21 : return ec;
333 : }
334 3786 : std::scoped_lock lock(mutex);
335 3786 : holds_session_cache = true;
336 3786 : return {};
337 3786 : }
338 :
339 : // Once, when open() fails or the listener stops: nothing if the manager is gone, and its
340 : // setting with it
341 27000 : void release_session_cache() {
342 : {
343 27000 : std::scoped_lock lock(mutex);
344 27000 : if (!std::exchange(holds_session_cache, false)) return;
345 27000 : }
346 3786 : if (auto tls = manager.lock()) tls->release_session_cache();
347 : }
348 :
349 : // Both executors run on an io_context: the sockets of this release need one
350 25258 : boost::system::error_code check_executors(const boost::asio::any_io_executor& accept_executor) const {
351 25258 : if (!io_context_of(accept_executor)) {
352 3 : LOG_ERROR("cannot listen on {}: the executor to accept on does not run on an io_context", url_for(configured_port()));
353 3 : return boost::asio::error::invalid_argument;
354 : }
355 25255 : if (!serve_on || !io_context_of(serve_on())) {
356 3 : LOG_ERROR("cannot listen on {}: the executor to serve on does not run on an io_context", url_for(configured_port()));
357 3 : return boost::asio::error::invalid_argument;
358 : }
359 25252 : return {};
360 : }
361 :
362 : template<typename Protocol>
363 25252 : boost::system::error_code open_on(const boost::asio::any_io_executor& executor, acceptor_thread_ptr thread,
364 : tls_setup connections_tls) {
365 25252 : typename Protocol::endpoint endpoint;
366 25252 : if (auto ec = endpoint_of(executor, endpoint)) return ec;
367 25246 : std::unique_ptr<typename Protocol::acceptor> acceptor;
368 25246 : if (auto ec = bind(executor, endpoint, acceptor)) return ec;
369 25225 : return publish<Protocol>(std::move(acceptor), std::move(thread), std::move(connections_tls));
370 25246 : }
371 :
372 13694 : boost::system::error_code endpoint_of(const boost::asio::any_io_executor& executor, tcp::endpoint& endpoint) const {
373 13694 : const auto& where = std::get<tcp_endpoint>(options.endpoint);
374 13694 : boost::system::error_code ec;
375 13694 : tcp::resolver resolver(executor);
376 13694 : auto results = resolver.resolve(where.host, std::to_string(where.port), ec);
377 13694 : if (!ec && results.empty()) ec = boost::asio::error::host_not_found;
378 13694 : if (ec) {
379 6 : LOG_ERROR("cannot resolve {}:{}: {}", where.host, where.port, ec.message());
380 6 : return ec;
381 : }
382 13688 : endpoint = results.begin()->endpoint();
383 13688 : return {};
384 13694 : }
385 :
386 11558 : boost::system::error_code endpoint_of(const boost::asio::any_io_executor&, local::endpoint& endpoint) const {
387 11558 : const auto& path = std::get<unix_endpoint>(options.endpoint).path;
388 11558 : if (path.empty()) {
389 0 : LOG_ERROR("cannot listen on a Unix socket without a path");
390 0 : return boost::asio::error::invalid_argument;
391 : }
392 : // A socket file left by an earlier server is replaced
393 11558 : std::error_code ec;
394 11558 : std::filesystem::remove(path, ec);
395 11558 : endpoint = local::endpoint(path);
396 11558 : return {};
397 : }
398 :
399 : // Bind and listen, retrying every LISTEN_RETRY_DELAY up to max_listening_attempts
400 : template<typename Endpoint, typename Acceptor>
401 25246 : boost::system::error_code bind(const boost::asio::any_io_executor& executor, const Endpoint& endpoint,
402 : std::unique_ptr<Acceptor>& acceptor) {
403 25246 : int attempts = 0;
404 6 : while (true) {
405 25246 : LOG_DEBUG("starting acceptor of {}", url_for(configured_port()));
406 25246 : auto ec = try_bind(executor, endpoint, acceptor);
407 25267 : if (!ec) return {};
408 21 : LOG_ERROR("cannot start listening on {}: {}", url_for(configured_port()), ec.message());
409 21 : if (options.max_listening_attempts >= 0 && ++attempts >= options.max_listening_attempts) return ec;
410 6 : std::unique_lock lock(mutex);
411 18 : if (changed.wait_for(lock, LISTEN_RETRY_DELAY, [this] { return stopping.load(); })) {
412 6 : return boost::asio::error::operation_aborted;
413 : }
414 : }
415 : }
416 :
417 : template<typename Endpoint, typename Acceptor>
418 25246 : static boost::system::error_code try_bind(const boost::asio::any_io_executor& executor, const Endpoint& endpoint,
419 : std::unique_ptr<Acceptor>& result) {
420 25246 : auto acceptor = std::make_unique<Acceptor>(executor);
421 25246 : boost::system::error_code ec;
422 25246 : acceptor->open(endpoint.protocol(), ec);
423 25246 : if (!ec) close_on_exec(acceptor->native_handle());
424 : if constexpr (std::is_same_v<Acceptor, tcp::acceptor>) {
425 13688 : if (!ec) acceptor->set_option(tcp::acceptor::reuse_address(true), ec);
426 : }
427 25246 : if (!ec) acceptor->bind(endpoint, ec);
428 25246 : if (!ec) acceptor->listen(boost::asio::socket_base::max_listen_connections, ec);
429 25246 : if (!ec) result = std::move(acceptor);
430 50492 : return ec;
431 25246 : }
432 :
433 : // Make the acceptor (and its thread) the listener's, unless it stopped meanwhile
434 : template<typename Protocol>
435 25225 : boost::system::error_code publish(std::unique_ptr<typename Protocol::acceptor> acceptor, acceptor_thread_ptr thread,
436 : tls_setup connections_tls) {
437 25225 : uint16_t bound = 0;
438 25225 : typename Protocol::endpoint local_endpoint;
439 25225 : boost::system::error_code ec;
440 25225 : local_endpoint = acceptor->local_endpoint(ec);
441 : if constexpr (std::is_same_v<Protocol, tcp>) {
442 13667 : if (!ec) bound = local_endpoint.port();
443 : } else {
444 11558 : set_permissions();
445 : }
446 : {
447 25225 : std::scoped_lock lock(mutex);
448 25225 : if (!stopping) {
449 : if constexpr (std::is_same_v<Protocol, tcp>) {
450 12719 : if (!ec) tcp_protocol = local_endpoint.protocol();
451 : }
452 23565 : acceptor_of<Protocol>() = std::move(acceptor);
453 23565 : acceptor_thread = std::move(thread);
454 23565 : tls = std::move(connections_tls);
455 23565 : port = bound;
456 23565 : url = url_for(bound);
457 23565 : opened = true;
458 : }
459 25225 : }
460 25225 : if (acceptor) {
461 : // Stopped while opening: its socket file goes first, as in close() (stop() removes it
462 : // once open only), and its acceptor before its thread
463 : if constexpr (std::is_same_v<Protocol, local>) {
464 712 : std::error_code removed;
465 712 : std::filesystem::remove(std::get<unix_endpoint>(options.endpoint).path, removed);
466 : }
467 1660 : acceptor.reset();
468 1660 : return boost::asio::error::operation_aborted;
469 : }
470 23565 : LOG_DEBUG("listening on {}", url_for(bound));
471 23565 : return {};
472 : }
473 :
474 : // Let the owner and its group use the socket file
475 11558 : void set_permissions() const {
476 11558 : const auto& path = std::get<unix_endpoint>(options.endpoint).path;
477 11558 : std::error_code ec;
478 11558 : std::filesystem::permissions(path,
479 : std::filesystem::perms::owner_all |
480 : std::filesystem::perms::group_read |
481 : std::filesystem::perms::group_write,
482 : std::filesystem::perm_options::replace, ec);
483 11558 : if (ec) LOG_WARNING("cannot set the permissions of Unix socket {}: {}", path, ec.message());
484 11558 : }
485 :
486 : // --- start() ---
487 :
488 23091 : void start(sync_callback<void(std::shared_ptr<socket>)> handler) {
489 : {
490 23091 : std::scoped_lock lock(mutex);
491 23091 : if (!opened || started || stopping || !handler) {
492 6 : LOG_ERROR("cannot start the listener on {}: {}", url_for(port),
493 : !handler ? "no handler" : !opened ? "not open" : started ? "started already" : "stopped");
494 6 : return;
495 : }
496 23085 : on_connection = std::move(handler);
497 23085 : started = true;
498 23085 : if (acceptor_thread) {
499 13158 : if (is_tcp()) accept_on_thread<tcp>();
500 8795 : else accept_on_thread<local>();
501 13158 : return;
502 : }
503 23091 : }
504 9927 : if (is_tcp()) accept_next<tcp>();
505 1881 : else accept_next<local>();
506 : }
507 :
508 : // With the mutex held: on a thread of its own, every pending connection is accepted at
509 : // once, without waiting
510 : template<typename Protocol>
511 13158 : void accept_on_thread() {
512 13158 : boost::system::error_code ec;
513 13158 : acceptor_of<Protocol>()->non_blocking(true, ec);
514 13158 : if (ec) {
515 0 : LOG_ERROR("cannot accept connections: {}", ec.message());
516 0 : return;
517 : }
518 39474 : boost::asio::post(acceptor_thread->get_io_context(), [self = shared_from_this()] {
519 13158 : self->accept_pending<Protocol>();
520 : });
521 : }
522 :
523 : // --- Accepting with executor providers ---
524 :
525 : // Without the mutex: accept the next connection on the next executor to serve on, unless
526 : // the listener stopped meanwhile (its acceptor is gone then)
527 : template<typename Protocol>
528 28442 : void accept_next() {
529 28442 : auto executor = serve_on();
530 28442 : bool usable = io_context_of(executor) != nullptr;
531 28442 : std::scoped_lock lock(mutex);
532 28442 : if (stopping) return;
533 28435 : auto& acceptor = acceptor_of<Protocol>();
534 28435 : if (!usable) executor = acceptor->get_executor();
535 80469 : acceptor->async_accept(executor, [self = shared_from_this(), usable]
536 : (const boost::system::error_code& e, typename Protocol::socket peer) {
537 26017 : self->on_accepted<Protocol>(e, std::move(peer), usable);
538 : });
539 28449 : }
540 :
541 : template<typename Protocol>
542 26017 : void on_accepted(const boost::system::error_code& e, typename Protocol::socket peer, bool usable) {
543 26017 : auto accepted_at = steady::now();
544 : // Where asio sets SO_NOSIGPIPE on the socket it accepted, this is a client already gone
545 : // (EINVAL by then): nothing wrong with the acceptor, and nothing to wait for
546 : #if defined(SO_NOSIGPIPE)
547 : bool client_gone = e == boost::asio::error::invalid_argument;
548 : #else
549 26017 : bool client_gone = false;
550 : #endif
551 : {
552 : // Stopped meanwhile (from any thread): its acceptor is gone
553 26017 : std::scoped_lock lock(mutex);
554 26017 : if (stopping) return;
555 18515 : if (e == boost::asio::error::operation_aborted) {
556 0 : LOG_DEBUG("stop accepting connections");
557 0 : return;
558 : }
559 18515 : if (e && !client_gone) {
560 0 : LOG_ERROR("cannot accept more connections: {}", e.message());
561 0 : retry_accept<Protocol>();
562 0 : return;
563 : }
564 26017 : }
565 18515 : if (client_gone) {
566 0 : accept_next<Protocol>();
567 0 : return;
568 : }
569 : // set before it reaches the executor serving it
570 18515 : close_on_exec(peer.native_handle());
571 18515 : accept_next<Protocol>();
572 18515 : if (!usable) {
573 0 : LOG_ERROR("cannot serve an accepted connection: the executor to serve on does not run on an io_context");
574 0 : return;
575 : }
576 37030 : serve_without_throwing([&] {
577 18515 : if (is_remote_allowed<Protocol>(peer.native_handle())) serve(std::move(peer), accepted_at);
578 : });
579 : }
580 :
581 : // --- Accepting on a thread of its own ---
582 :
583 : template<typename Protocol>
584 27636 : void wait_connections() {
585 55272 : acceptor_of<Protocol>()->async_wait(boost::asio::socket_base::wait_read,
586 72812 : [self = shared_from_this()](const boost::system::error_code& e) {
587 17540 : if (!e) {
588 17540 : self->accept_pending<Protocol>();
589 0 : } else if (e != boost::asio::error::operation_aborted) {
590 0 : LOG_ERROR("cannot accept more connections: {}", e.message());
591 0 : self->retry_accept<Protocol>();
592 : } else {
593 0 : LOG_DEBUG("stop accepting connections");
594 : }
595 : });
596 27636 : }
597 :
598 : // Only the acceptor thread uses the acceptor (stop() joins it before closing the acceptor),
599 : // and it is the only one running its io_context, so no other thread takes its readiness
600 : template<typename Protocol>
601 30698 : void accept_pending() {
602 30698 : auto& acceptor = acceptor_of<Protocol>();
603 47954 : while (!stopping) {
604 : #if defined(__linux__)
605 44892 : int descriptor = ::accept4(acceptor->native_handle(), nullptr, nullptr, SOCK_CLOEXEC);
606 : #else
607 : int descriptor = ::accept(acceptor->native_handle(), nullptr, nullptr);
608 : if (descriptor >= 0) close_on_exec(descriptor);
609 : #endif
610 44892 : if (descriptor >= 0) {
611 17256 : accepted_descriptor accepted(descriptor);
612 17256 : auto accepted_at = steady::now();
613 : #if defined(SO_NOSIGPIPE)
614 : int enabled = 1;
615 : ::setsockopt(descriptor, SOL_SOCKET, SO_NOSIGPIPE, &enabled, sizeof(enabled));
616 : #endif
617 34512 : serve_without_throwing([&] { hand_over<Protocol>(std::move(accepted), accepted_at); });
618 17256 : continue;
619 17256 : }
620 27636 : int error = errno;
621 27636 : if (error == EINTR || error == ECONNABORTED || error == EPROTO) continue;
622 27636 : if (error == EAGAIN || error == EWOULDBLOCK) {
623 27636 : wait_connections<Protocol>();
624 : } else {
625 0 : LOG_ERROR("cannot accept more connections: {}", std::error_code(error, std::system_category()).message());
626 0 : retry_accept<Protocol>();
627 : }
628 27636 : return;
629 : }
630 : }
631 :
632 : // The connection is registered and set up on the executor serving it, waking it up once
633 : template<typename Protocol>
634 17256 : void hand_over(accepted_descriptor descriptor, steady::time_point accepted_at) {
635 17256 : if (!is_remote_allowed<Protocol>(descriptor.get())) return;
636 17250 : auto executor = serve_on();
637 17250 : if (!io_context_of(executor)) {
638 0 : LOG_ERROR("cannot serve an accepted connection: the executor to serve on does not run on an io_context");
639 0 : return;
640 : }
641 17250 : auto protocol = protocol_of<Protocol>();
642 17250 : boost::asio::post(executor,
643 34500 : [self = shared_from_this(), executor, protocol, descriptor = std::move(descriptor), accepted_at]() mutable {
644 17250 : typename Protocol::socket peer(executor);
645 17250 : boost::system::error_code ec;
646 17250 : peer.assign(protocol, descriptor.get(), ec);
647 17250 : if (ec) {
648 0 : LOG_ERROR("cannot serve an accepted connection: {}", ec.message());
649 0 : return;
650 : }
651 17250 : descriptor.release();
652 34500 : serve_without_throwing([&] { self->serve(std::move(peer), accepted_at); });
653 17250 : });
654 17250 : }
655 :
656 : template<typename Protocol>
657 17250 : Protocol protocol_of() const {
658 3950 : if constexpr (std::is_same_v<Protocol, tcp>) return tcp_protocol;
659 13300 : else return local();
660 : }
661 :
662 : // Accept again after a delay: an error accepting (e.g. out of descriptors) would repeat
663 : // at once otherwise
664 : template<typename Protocol>
665 0 : void retry_accept() {
666 0 : auto timer = std::make_shared<boost::asio::steady_timer>(acceptor_of<Protocol>()->get_executor(),
667 : ACCEPT_RETRY_DELAY);
668 0 : timer->async_wait([self = shared_from_this(), timer](const boost::system::error_code& e) {
669 0 : if (e) return;
670 0 : if (self->own_acceptor_thread) self->accept_pending<Protocol>();
671 0 : else self->accept_next<Protocol>();
672 : });
673 0 : }
674 :
675 : // Whether to serve a connection just accepted, for its remote address (TCP only)
676 : template<typename Protocol>
677 35771 : bool is_remote_allowed(int descriptor) const {
678 : if constexpr (!std::is_same_v<Protocol, tcp>) {
679 16767 : return true;
680 : } else {
681 : // The remote address is only looked up to filter it
682 19004 : if (allowed_remotes.empty() && forbidden_remotes.empty()) return true;
683 18 : tcp::endpoint remote;
684 18 : auto size = static_cast<socklen_t>(remote.capacity());
685 18 : std::string remote_ip = "0.0.0.0";
686 18 : if (::getpeername(descriptor, remote.data(), &size) == 0) {
687 18 : remote.resize(size);
688 18 : remote_ip = remote.address().to_string();
689 : }
690 42 : if (!forbidden_remotes.contains(remote_ip) &&
691 24 : (allowed_remotes.empty() || allowed_remotes.contains(remote_ip))) {
692 6 : return true;
693 : }
694 24 : LOG_DEBUG("rejecting connection from: ip: {}, port: {}, secure: {}",
695 : remote_ip, port.load(), secure);
696 12 : return false;
697 18 : }
698 : }
699 :
700 : // --- Serving ---
701 :
702 : // A connection accepted, on the executor serving it (see serve_without_throwing())
703 18992 : void serve(tcp::socket peer, steady::time_point accepted_at) {
704 18992 : auto sock = make_socket(std::move(peer));
705 18992 : LOG_DEBUG("received connection from: ip: {}, port: {}, secure: {}",
706 : sock->get_remote_ip(), sock->get_local_port(), sock->is_secure());
707 : // Fails if the connection is already gone: drop it
708 18992 : if (options.tcp_no_delay && !sock->enable_tcp_no_delay()) return;
709 18992 : if (sock->requires_handshake()) {
710 8292 : handshake(std::move(sock), accepted_at);
711 : } else {
712 10700 : deliver(std::move(sock));
713 : }
714 18992 : }
715 :
716 : // A TLS connection is created with the context of the listener and asks for client
717 : // certificates with its policy (the mode of the host name, if it has one, replaces it in the
718 : // handshake: see certificate_manager::sni_callback())
719 18992 : std::shared_ptr<tcp_socket> make_socket(tcp::socket peer) const {
720 18992 : if (!tls.context) return std::make_shared<tcp_socket>("listener", std::move(peer));
721 : // On an io_context: checked as it was accepted
722 8292 : auto* io_context = io_context_of(peer.get_executor());
723 8292 : auto sock = std::make_shared<ssl_socket>("listener", *io_context, tls.context);
724 8292 : sock->get_socket() = std::move(peer);
725 8292 : sock->set_client_verify(tls.client_verify);
726 8292 : return sock;
727 8292 : }
728 :
729 : // --- TLS handshakes ---
730 :
731 : // The end of the handshake of a connection accepted at `accepted_at`: none without a limit
732 8274 : std::optional<steady::time_point> handshake_deadline(steady::time_point accepted_at) const {
733 8274 : if (options.handshake_timeout.count() <= 0) return std::nullopt;
734 8265 : return accepted_at + options.handshake_timeout;
735 : }
736 :
737 : // Start the handshake of a TLS connection, registered as pending until it ends: on the
738 : // io_context serving it, or on the strand of its gate with a bound
739 8292 : void handshake(std::shared_ptr<tcp_socket> sock, steady::time_point accepted_at) {
740 8292 : pending_handshake entry{sock, sock->get_io_context().get_executor(), nullptr};
741 : {
742 8292 : std::scoped_lock lock(mutex);
743 8292 : if (stopping) {
744 18 : sock->close();
745 18 : return;
746 : }
747 8274 : if (options.max_pending_handshakes > 0) {
748 129 : entry.gate = gate_of(sock->get_io_context());
749 129 : entry.executor = entry.gate->strand;
750 : }
751 8274 : pending.emplace(sock.get(), entry);
752 8292 : }
753 8274 : co_spawn(entry.executor, serve_tls(shared_from_this(), std::move(sock), entry.gate, accepted_at),
754 : log_serve_failure);
755 8292 : }
756 :
757 : // With the mutex held
758 129 : std::shared_ptr<handshake_gate> gate_of(boost::asio::io_context& io_context) {
759 129 : auto& gate = gates[&io_context];
760 129 : if (!gate) gate = std::make_shared<handshake_gate>(io_context);
761 129 : return gate;
762 : }
763 :
764 : // The handshake of a TLS connection, then the handler, unless the listener stopped meanwhile
765 8274 : static awaitable<void> serve_tls(std::shared_ptr<state> self, std::shared_ptr<tcp_socket> sock,
766 : std::shared_ptr<handshake_gate> gate, steady::time_point accepted_at) {
767 : auto deadline = self->handshake_deadline(accepted_at);
768 : boost::system::error_code ec;
769 : {
770 : pending_registration registered(self, sock.get());
771 : if (gate) {
772 : ec = co_await handshake_in_slot(self, sock, std::move(gate), deadline);
773 : } else {
774 : ec = co_await handshake_until(sock, deadline);
775 : }
776 : }
777 : // Not errors of the listener: any client (or scanner) may cause them
778 : if (ec == boost::asio::error::timed_out) {
779 : LOG_DEBUG("TLS handshake not done in time, remote ip: {}", sock->get_remote_ip());
780 : sock->close();
781 : } else if (ec && self->stopping) {
782 : sock->close();
783 : } else if (ec) {
784 : LOG_DEBUG("error while handling SSL handshake: {}, remote ip: {}", ec.message(), sock->get_remote_ip());
785 : } else {
786 : self->deliver(std::move(sock));
787 : }
788 16548 : }
789 :
790 : // Removes a connection from the pending ones once its handshake ended, however it ended
791 : struct pending_registration {
792 : std::shared_ptr<state> self;
793 : const tcp_socket* sock;
794 :
795 8274 : pending_registration(std::shared_ptr<state> listener_state, const tcp_socket* pending_sock)
796 8274 : : self(std::move(listener_state)), sock(pending_sock) {}
797 : pending_registration(const pending_registration&) = delete;
798 : pending_registration& operator=(const pending_registration&) = delete;
799 8274 : ~pending_registration() {
800 8274 : std::scoped_lock lock(self->mutex);
801 8274 : self->pending.erase(sock);
802 8274 : }
803 : };
804 :
805 : // On the strand of the gate: the handshake in a slot, waited for within the deadline
806 129 : static awaitable<boost::system::error_code> handshake_in_slot(std::shared_ptr<state> self,
807 : std::shared_ptr<tcp_socket> sock,
808 : std::shared_ptr<handshake_gate> gate,
809 : std::optional<steady::time_point> deadline) {
810 : if (auto ec = co_await wait_for_slot(self, *gate, deadline)) co_return ec;
811 : handshake_slot slot(std::move(gate));
812 : co_return co_await handshake_until(std::move(sock), deadline);
813 258 : }
814 :
815 : // On the strand of the gate: until a slot is free (taken by the caller before it suspends
816 : // again), `timed_out` at the deadline, or `operation_aborted` once the listener stops
817 129 : static awaitable<boost::system::error_code> wait_for_slot(std::shared_ptr<state> self, handshake_gate& gate,
818 : std::optional<steady::time_point> deadline) {
819 : while (true) {
820 : if (self->stopping) co_return boost::asio::error::operation_aborted;
821 : if (gate.in_progress < self->options.max_pending_handshakes) co_return boost::system::error_code{};
822 : // Set again by the next slot freed, or by stop()
823 : gate.slot_freed.reset();
824 : if (!deadline) {
825 : co_await gate.slot_freed.wait();
826 : continue;
827 : }
828 : auto left = *deadline - steady::now();
829 : if (left <= steady::duration::zero() || !co_await with_timeout(gate.slot_freed.wait(), left)) {
830 : co_return boost::asio::error::timed_out;
831 : }
832 : }
833 258 : }
834 :
835 16767 : void serve(local::socket peer, steady::time_point) {
836 16767 : auto sock = std::make_shared<unix_socket>("listener", std::move(peer));
837 16767 : LOG_DEBUG("received connection on Unix socket: {}", std::get<unix_endpoint>(options.endpoint).path);
838 16767 : deliver(std::move(sock));
839 16767 : }
840 :
841 : // Hand a connection to the handler, unless the listener is stopping (see above)
842 32042 : void deliver(std::shared_ptr<socket> sock) {
843 32042 : auto delivering_thread = std::this_thread::get_id();
844 : {
845 32042 : std::scoped_lock lock(mutex);
846 32042 : if (stopping) {
847 4246 : sock->close();
848 4246 : return;
849 : }
850 27796 : delivering.push_back(delivering_thread);
851 32042 : }
852 55592 : serve_without_throwing([&] { on_connection(std::move(sock)); });
853 : {
854 : // Not there if the handler stopped the listener
855 27796 : std::scoped_lock lock(mutex);
856 27796 : auto it = std::find(delivering.begin(), delivering.end(), delivering_thread);
857 27796 : if (it != delivering.end()) delivering.erase(it);
858 27796 : }
859 27796 : changed.notify_all();
860 : }
861 :
862 : // --- stop() ---
863 :
864 50497 : void stop() {
865 50497 : std::unique_lock lock(mutex);
866 50497 : bool first = !stopping;
867 50497 : stopping = true;
868 : // A handler stopping the listener does not wait for itself
869 50497 : std::erase(delivering, std::this_thread::get_id());
870 50497 : if (first) {
871 25307 : auto thread = std::move(acceptor_thread);
872 25307 : auto handshakes = std::exchange(pending, {});
873 : // Their strands go with the handshakes still holding them, not with this state
874 25307 : auto released_gates = std::exchange(gates, {});
875 25307 : auto released_manager = std::exchange(certificates, nullptr);
876 25307 : lock.unlock();
877 25307 : changed.notify_all(); // an open() waiting to bind again gives up
878 25307 : close(std::move(thread));
879 25307 : end_handshakes(handshakes);
880 25307 : released_gates.clear();
881 25307 : release_session_cache();
882 : // Destroyed here if nothing else holds it, outside the lock: the handshakes still
883 : // running find it gone (see certificate_manager::sni_callback())
884 25307 : released_manager.reset();
885 25307 : lock.lock();
886 25307 : }
887 101047 : changed.wait(lock, [this] { return delivering.empty(); });
888 50497 : }
889 :
890 : // Close the connections whose handshake is pending, on the executor they handshake on, and
891 : // wake those waiting for a slot: they see the listener stopping and end, without serving.
892 : // Those gone already (their io_context with them, maybe) are left alone
893 25307 : static void end_handshakes(const std::unordered_map<const tcp_socket*, pending_handshake>& handshakes) {
894 25838 : for (const auto& [_, handshake] : handshakes) {
895 531 : if (handshake.sock.expired()) continue;
896 520 : boost::asio::post(handshake.executor, [weak = handshake.sock, gate = handshake.gate] {
897 440 : if (auto sock = weak.lock()) sock->close();
898 440 : if (gate) gate->slot_freed.set();
899 440 : });
900 : }
901 25307 : }
902 :
903 : // Close the acceptor, once its thread (if any) is not using it. The Unix socket file goes
904 : // first, as in libuv and Go: a server binding its path as soon as a client cannot connect
905 : // does not lose its new file to this one. The accept handlers still to run do nothing, as
906 : // the listener is stopping
907 25307 : void close(acceptor_thread_ptr thread) {
908 25307 : if (thread) thread->stop();
909 25307 : std::unique_ptr<tcp::acceptor> tcp_closed;
910 25307 : std::unique_ptr<local::acceptor> unix_closed;
911 25307 : bool was_open = false;
912 : {
913 25307 : std::scoped_lock lock(mutex);
914 25307 : tcp_closed = std::move(tcp_acceptor);
915 25307 : unix_closed = std::move(unix_acceptor);
916 25307 : was_open = opened;
917 25307 : }
918 25307 : if (const auto* unix = std::get_if<unix_endpoint>(&options.endpoint); unix && was_open) {
919 10846 : std::error_code ec;
920 10846 : std::filesystem::remove(unix->path, ec);
921 10846 : if (ec) LOG_WARNING("cannot remove Unix socket file {}: {}", unix->path, ec.message());
922 : }
923 25307 : close_acceptor(tcp_closed);
924 25307 : close_acceptor(unix_closed);
925 25307 : port = 0;
926 : // The thread goes after its acceptor, with the handlers it never ran
927 25307 : tcp_closed.reset();
928 25307 : unix_closed.reset();
929 25307 : }
930 :
931 : template<typename Acceptor>
932 50614 : static void close_acceptor(const std::unique_ptr<Acceptor>& acceptor) {
933 50614 : if (!acceptor || !acceptor->is_open()) return;
934 23565 : boost::system::error_code ec;
935 23565 : acceptor->close(ec);
936 23565 : if (ec) LOG_WARNING("cannot close the acceptor: {}", ec.message());
937 : }
938 : };
939 :
940 25307 : listener::listener(listener_options options, executor_provider accept_on, executor_provider serve_on)
941 25307 : : state_(std::make_shared<state>(std::move(options), std::move(accept_on), std::move(serve_on))) {}
942 :
943 25307 : listener::~listener() {
944 : try {
945 25307 : stop();
946 0 : } catch (const std::exception& e) {
947 0 : LOG_ERROR("cannot stop a listener: {}", e.what());
948 0 : }
949 25307 : }
950 :
951 25310 : boost::system::error_code listener::open() {
952 25310 : return state_->open();
953 : }
954 :
955 23091 : void listener::start(sync_callback<void(std::shared_ptr<socket>)> on_connection) {
956 23091 : state_->start(std::move(on_connection));
957 23091 : }
958 :
959 50497 : void listener::stop() {
960 50497 : state_->stop();
961 50497 : }
962 :
963 26040 : uint16_t listener::local_port() const {
964 26040 : return state_->port;
965 : }
966 :
967 21255 : std::string listener::url() const {
968 21255 : std::scoped_lock lock(state_->mutex);
969 42510 : return state_->url;
970 21255 : }
971 :
972 15245 : void detail::listener_access::use_acceptor_thread(listener& l) {
973 15245 : std::scoped_lock lock(l.state_->mutex);
974 15245 : if (!l.state_->opened) l.state_->own_acceptor_thread = true;
975 15245 : }
976 :
977 : } // namespace thinger::asio
|
