Line data Source code
1 : #include "http_application.hpp"
2 : #include "request.hpp"
3 : #include "response.hpp"
4 : #include "memory_response.hpp"
5 : #include "../../asio/async_event.hpp"
6 : #include "../../util/logger.hpp"
7 : #include "../../util/base64.hpp"
8 : #include "../../util/ascii.hpp"
9 : #include <boost/algorithm/string.hpp>
10 : #include <boost/intrusive/list.hpp>
11 : #include <atomic>
12 : #include <mutex>
13 : #include <optional>
14 : #include <utility>
15 : #include <vector>
16 :
17 : namespace thinger::http {
18 :
19 : // What serving a request reads: the hosts and their routes, the middlewares and the settings of
20 : // the application. Its connections and its dispatches in memory keep it (see connection_handler()
21 : // and dispatch()), so the requests still running when the application is destroyed find it. It
22 : // never refers to the application
23 : class http_application::state : public std::enable_shared_from_this<state> {
24 : public:
25 11193 : explicit state(std::unique_ptr<virtual_host> default_host) : default_host_(std::move(default_host)) {}
26 :
27 : // The source of a request dispatched in memory, which is also how close() reaches it: it
28 : // lives as the request does, as long as a handler outliving its answer may run. The request
29 : // owns it, and it does not own the request, nor what reaches it through a response hook
30 : // (the response is only weakly held)
31 : struct tracked_source : detail::memory_source, boost::intrusive::list_base_hook<> {
32 4077 : tracked_source(std::string body, std::string peer_ip, std::shared_ptr<state> owner,
33 : std::shared_ptr<http_stream> stream, std::weak_ptr<memory_response> memory)
34 8154 : : detail::memory_source(std::move(body), std::move(peer_ip)), owner(std::move(owner)),
35 12231 : stream(std::move(stream)), memory(std::move(memory)) {}
36 : ~tracked_source() override;
37 :
38 : std::shared_ptr<state> owner;
39 : std::shared_ptr<http_stream> stream;
40 : std::weak_ptr<memory_response> memory;
41 : };
42 :
43 : // In-memory dispatch, on a frame keeping this state instead of the application. Static: a
44 : // member coroutine would only keep a raw pointer to it
45 : static awaitable<std::shared_ptr<http_response>> dispatch(std::shared_ptr<state> self,
46 : std::shared_ptr<http_request> http_request,
47 : dispatch_options options);
48 : // What an in-memory dispatch collects once its timeout passed without a response: `res`,
49 : // the response if created by then. Static, as dispatch()
50 : static awaitable<std::shared_ptr<http_response>> answer_dispatch_timeout(std::shared_ptr<state> self,
51 : std::shared_ptr<memory_response> memory,
52 : std::shared_ptr<http_stream> stream,
53 : std::optional<response> res);
54 : // Track an in-memory dispatch; false if the application is gone already
55 : bool track(tracked_source& source);
56 : // The application is being destroyed: from now on no request starts, and the dispatches in
57 : // flight are ended and their handlers interrupted, without waiting for them
58 : void close();
59 34492 : bool closed() const { return closed_.load(std::memory_order_acquire); }
60 :
61 : // handle(), telling `created` the response once created (for in-memory dispatch, which
62 : // answers its timeout)
63 : awaitable<void> handle(std::shared_ptr<request> req, std::shared_ptr<response_sink> sink,
64 : std::function<void(const response&)> created);
65 : virtual_host& resolve_host(request& req);
66 : std::shared_ptr<http_response> make_error_response(http_response::status status, const std::string& message) const;
67 : // Middlewares, body reading and handler of a matched request, within the cancellation scope
68 : // of its stream (on a coroutine of their own if none is bound to it, see
69 : // http_stream::bind_handler()), then what is left of its body
70 : awaitable<void> serve(std::shared_ptr<request> req, response& res, virtual_host& host, const route* matched_route);
71 : // The same, within the timeout of the route (or of the host)
72 : awaitable<void> serve_with_timeout(std::shared_ptr<request> req, response res, virtual_host& host,
73 : const route* matched_route, std::chrono::milliseconds timeout,
74 : http_response::status status);
75 : awaitable<bool> run_middlewares(request& req, response& res);
76 : awaitable<void> discard_unread_body(request& req);
77 : // What CORS means for the request, set on its response before anything may answer it, and
78 : // what comes next
79 : enum class cors_step { go_on, answer_preflight, fail };
80 : cors_step decide_cors(const http_request& head, response& res) const;
81 : // Answers a request that CORS alone settles: a preflight, or an origin it could not decide
82 : awaitable<void> answer_cors(request& req, response& res, cors_step step);
83 : // Whether a handler can start, marking it as started if so
84 : bool start_handler(http_stream* stream) const;
85 :
86 : // The application as the default host: its settings, shared with the application
87 : std::unique_ptr<virtual_host> default_host_;
88 :
89 : // Body decoders, accepted types and limits (before openapi_, which documents them)
90 : body_settings body_settings_;
91 : openapi_generator openapi_{default_host_->router(), &body_settings_};
92 : std::vector<async_middleware_function> middlewares_;
93 :
94 : // Cross-origin requests (off if null)
95 : std::shared_ptr<const detail::cors_policy> cors_;
96 :
97 : // Maximum allowed request body size
98 : std::uint64_t max_body_size_{default_max_body_size};
99 :
100 : // Virtual hosts by exact name, and by pattern (in registration order)
101 : std::map<std::string, std::unique_ptr<virtual_host>> exact_hosts_;
102 : std::vector<std::unique_ptr<virtual_host>> pattern_hosts_;
103 :
104 : // Body of error responses (default format if null)
105 : std::shared_ptr<const error_formatter> error_formatter_;
106 :
107 : // Proxies allowed to report the client address (none if null)
108 : std::shared_ptr<const trusted_proxies> trusted_proxies_;
109 :
110 : // Called with every response (none if null)
111 : std::shared_ptr<const response_hook> response_hook_;
112 :
113 : // The dispatches in flight (see close()). Written under the mutex, closed_ is read without it
114 : std::mutex dispatches_mutex_;
115 : boost::intrusive::list<tracked_source> dispatches_;
116 : std::atomic<bool> closed_{false};
117 : };
118 :
119 11193 : http_application::http_application()
120 11193 : : state_(std::make_shared<state>(std::unique_ptr<virtual_host>(new virtual_host(settings_)))) {}
121 :
122 11754 : http_application::~http_application() {
123 11193 : state_->close();
124 11754 : }
125 :
126 4077 : bool http_application::state::track(tracked_source& source) {
127 4077 : std::scoped_lock lock(dispatches_mutex_);
128 4077 : if (closed_.load(std::memory_order_relaxed)) return false;
129 4059 : dispatches_.push_back(source);
130 4059 : return true;
131 4077 : }
132 :
133 8154 : http_application::state::tracked_source::~tracked_source() {
134 4077 : std::scoped_lock lock(owner->dispatches_mutex_);
135 4077 : if (is_linked()) owner->dispatches_.erase(owner->dispatches_.iterator_to(*this));
136 8154 : }
137 :
138 11193 : void http_application::state::close() {
139 11193 : std::vector<std::pair<std::shared_ptr<http_stream>, std::weak_ptr<memory_response>>> in_flight;
140 : {
141 11193 : std::scoped_lock lock(dispatches_mutex_);
142 11193 : closed_.store(true, std::memory_order_release);
143 11193 : in_flight.reserve(dispatches_.size());
144 13396 : for (const auto& source : dispatches_) in_flight.emplace_back(source.stream, source.memory);
145 11193 : }
146 13396 : for (const auto& [stream, memory] : in_flight) {
147 : // First the response, so what an interrupted handler answers is dropped; one already
148 : // complete is delivered
149 2203 : if (auto response = memory.lock()) response->abandon();
150 2203 : stream->cancel(cancel_reason::shutdown);
151 : }
152 11193 : }
153 :
154 23786 : std::function<awaitable<void>(std::shared_ptr<request>)> http_application::connection_handler() const {
155 47572 : return [state = state_](std::shared_ptr<request> req) {
156 23503 : auto sink = std::make_shared<connection_sink>(*req);
157 47006 : return state->handle(std::move(req), std::move(sink), nullptr);
158 71075 : };
159 : }
160 :
161 : // Middleware
162 2092 : void http_application::use(async_middleware_function middleware) {
163 2092 : state_->middlewares_.push_back(std::move(middleware));
164 2092 : }
165 :
166 921 : void http_application::use(middleware_function middleware) {
167 2508 : use([middleware = std::move(middleware)](request& req, response& res) -> awaitable<bool> {
168 : // shared flag: a misbehaving middleware may still call next() after returning
169 : auto passed = std::make_shared<bool>(false);
170 1516 : middleware(req, res, [passed]() { *passed = true; });
171 : co_return *passed;
172 3174 : });
173 921 : }
174 :
175 39 : void http_application::use(coroutine_middleware_function middleware) {
176 141 : use([middleware = std::move(middleware)](request& req, response& res) -> awaitable<bool> {
177 : auto passed = std::make_shared<bool>(false);
178 99 : co_await middleware(req, res, [passed]() { *passed = true; });
179 : co_return *passed;
180 204 : });
181 39 : }
182 :
183 : // Basic Auth helpers
184 64 : void http_application::set_basic_auth(const std::string& path_prefix,
185 : const std::string& realm,
186 : auth_verify_function verify) {
187 146 : use([path_prefix, realm, verify](request& req, response& res) -> awaitable<bool> {
188 : // Get the request path
189 : auto http_request = req.get_http_request();
190 : if (!http_request) co_return true;
191 :
192 : const std::string& path = http_request->get_uri();
193 :
194 : // Check if this path requires auth
195 : if (!path.starts_with(path_prefix)) co_return true;
196 :
197 : // Check for Authorization header
198 : if (!http_request->has_header("Authorization")) {
199 : res.header("WWW-Authenticate", "Basic realm=\"" + realm + "\"");
200 : res.error(http_response::status::unauthorized, "Authentication required");
201 : co_return false;
202 : }
203 :
204 : auto auth_header = http_request->get_header("Authorization");
205 : if (!auth_header.starts_with("Basic ")) {
206 : res.header("WWW-Authenticate", "Basic realm=\"" + realm + "\"");
207 : res.error(http_response::status::unauthorized, "Invalid authentication");
208 : co_return false;
209 : }
210 :
211 : // Decode base64 credentials
212 : auto encoded = auth_header.substr(6);
213 : std::string decoded;
214 : try {
215 : decoded = ::thinger::util::base64::decode(encoded);
216 : } catch (...) {
217 : res.error(http_response::status::unauthorized, "Invalid credentials format");
218 : co_return false;
219 : }
220 :
221 : // Parse username:password
222 : auto colon_pos = decoded.find(':');
223 : if (colon_pos == std::string::npos) {
224 : res.error(http_response::status::unauthorized, "Invalid credentials format");
225 : co_return false;
226 : }
227 :
228 : std::string username = decoded.substr(0, colon_pos);
229 : std::string password = decoded.substr(colon_pos + 1);
230 :
231 : // Verify credentials
232 : if (verify(username, password)) {
233 : req.set_auth_user(username);
234 : co_return true;
235 : }
236 :
237 : res.header("WWW-Authenticate", "Basic realm=\"" + realm + "\"");
238 : res.error(http_response::status::unauthorized, "Invalid username or password");
239 : co_return false;
240 164 : });
241 64 : }
242 :
243 54 : void http_application::set_basic_auth(const std::string& path_prefix,
244 : const std::string& realm,
245 : const std::string& username,
246 : const std::string& password) {
247 54 : set_basic_auth(path_prefix, realm,
248 108 : [username, password](const std::string& u, const std::string& p) {
249 4 : return u == username && p == password;
250 : });
251 54 : }
252 :
253 6 : void http_application::set_basic_auth(const std::string& path_prefix,
254 : const std::string& realm,
255 : const std::map<std::string, std::string>& users) {
256 6 : set_basic_auth(path_prefix, realm,
257 12 : [users](const std::string& u, const std::string& p) {
258 6 : auto it = users.find(u);
259 6 : return it != users.end() && it->second == p;
260 : });
261 6 : }
262 :
263 : // Virtual hosts
264 252 : virtual_host& http_application::host(const std::string& name) {
265 252 : if (name.empty() || name == "*") return *this;
266 :
267 249 : if (!virtual_host::is_host_pattern(name)) {
268 204 : auto key = virtual_host::normalize_host(name);
269 204 : auto& host = state_->exact_hosts_[key];
270 204 : if (!host) host = std::make_unique<virtual_host>(key);
271 204 : return *host;
272 204 : }
273 :
274 : // Registering the same pattern again returns the existing host
275 45 : auto pattern = name;
276 45 : ::thinger::util::ascii::to_lower(pattern);
277 57 : for (const auto& host : state_->pattern_hosts_) {
278 18 : if (host->name() == pattern) return *host;
279 : }
280 39 : return *state_->pattern_hosts_.emplace_back(std::make_unique<virtual_host>(pattern));
281 45 : }
282 :
283 24 : virtual_host& http_application::host_regex(const std::string& pattern) {
284 48 : return *state_->pattern_hosts_.emplace_back(
285 48 : std::make_unique<virtual_host>(pattern, std::regex(pattern, std::regex::icase)));
286 : }
287 :
288 27136 : virtual_host& http_application::state::resolve_host(request& req) {
289 : // Without virtual hosts, every request is served by the application
290 27136 : if (exact_hosts_.empty() && pattern_hosts_.empty()) {
291 26884 : req.set_virtual_host(default_host_.get());
292 26884 : return *default_host_;
293 : }
294 :
295 252 : auto http_request = req.get_http_request();
296 252 : const auto& host_header = http_request->get_header(header::host);
297 252 : auto name = virtual_host::normalize_host(host_header.empty() ? http_request->get_host() : host_header);
298 :
299 : // A name longer than any DNS name (RFC 1035 section 2.3.4: 253 bytes without the trailing
300 : // dot) cannot be one of a pattern host, and is not run through its regex: libstdc++ matches
301 : // recursively, about one frame per byte, so a Host of kilobytes could overflow the stack.
302 : // It is served by the default host, as any other name no virtual host matches
303 252 : constexpr size_t max_host_name_size = 253;
304 :
305 252 : virtual_host* host = default_host_.get();
306 252 : if (auto it = exact_hosts_.find(name); it != exact_hosts_.end()) {
307 120 : host = it->second.get();
308 132 : } else if (name.size() <= max_host_name_size) {
309 165 : for (const auto& candidate : pattern_hosts_) {
310 75 : std::smatch matches;
311 75 : if (!candidate->matches(name, matches)) continue;
312 :
313 27 : std::vector<std::string> captures;
314 81 : for (size_t i = 0; i < matches.size(); ++i) {
315 54 : captures.push_back(matches[i].str());
316 54 : const auto& parameters = candidate->get_parameters();
317 54 : if (i > 0 && i <= parameters.size() && !parameters[i - 1].empty() && matches[i].matched) {
318 3 : req.set_uri_parameter(parameters[i - 1], matches[i].str());
319 : }
320 : }
321 27 : req.set_host_matches(std::move(captures));
322 27 : host = candidate.get();
323 27 : break;
324 102 : }
325 : }
326 :
327 252 : req.set_virtual_host(host);
328 252 : return *host;
329 252 : }
330 :
331 : // Error format
332 90 : void http_application::set_error_formatter(error_formatter formatter) {
333 90 : state_->error_formatter_ = formatter ? std::make_shared<const error_formatter>(std::move(formatter)) : nullptr;
334 90 : }
335 :
336 480 : std::shared_ptr<http_response> http_application::state::make_error_response(http_response::status status,
337 : const std::string& message) const {
338 480 : auto result = std::make_shared<http_response>();
339 480 : result->set_status(status);
340 960 : format_error(error_formatter_.get(), {status, message, nullptr}, *result);
341 480 : return result;
342 480 : }
343 :
344 : // Trusted proxies
345 39 : bool http_application::set_trusted_proxies(const std::vector<std::string>& proxies, forwarded_header header) {
346 39 : auto trusted = std::make_shared<trusted_proxies>(header);
347 96 : for (const auto& proxy : proxies) {
348 60 : if (!trusted->add(proxy)) {
349 3 : LOG_ERROR("Invalid trusted proxy: '{}'", proxy);
350 3 : state_->trusted_proxies_.reset();
351 3 : return false;
352 : }
353 : }
354 36 : state_->trusted_proxies_ = trusted->empty() ? nullptr : std::move(trusted);
355 36 : return true;
356 39 : }
357 :
358 : // Response hook
359 1074 : void http_application::on_response(response_hook hook) {
360 1074 : state_->response_hook_ = hook ? std::make_shared<const response_hook>(std::move(hook)) : nullptr;
361 1074 : }
362 :
363 : // OpenAPI
364 16 : route& http_application::serve_openapi(const std::string& path) {
365 : // The generator outlives the route: both are the state's
366 32 : return get(path, [openapi = &state_->openapi_](response& res) {
367 2 : res.json(openapi->generate());
368 34 : }).hidden();
369 : }
370 :
371 61 : openapi_generator& http_application::openapi() {
372 61 : return state_->openapi_;
373 : }
374 :
375 : // Configuration
376 575 : boost::system::error_code http_application::enable_cors(cors_options options) {
377 575 : auto policy = detail::make_cors_policy(std::move(options));
378 575 : if (!policy) return make_error_code(boost::system::errc::invalid_argument);
379 392 : state_->cors_ = std::move(policy);
380 392 : return {};
381 575 : }
382 :
383 12 : void http_application::disable_cors() {
384 12 : state_->cors_.reset();
385 12 : }
386 :
387 733 : void http_application::set_max_body_size(std::uint64_t size) {
388 733 : state_->max_body_size_ = size;
389 733 : }
390 :
391 0 : std::uint64_t http_application::get_max_body_size() const {
392 0 : return state_->max_body_size_;
393 : }
394 :
395 6 : body_decoder_registry& http_application::body_decoders() {
396 6 : return state_->body_settings_.decoders;
397 : }
398 :
399 3 : const std::vector<std::string>& http_application::get_body_types() const {
400 3 : return state_->body_settings_.accepted;
401 : }
402 :
403 3 : void http_application::set_body_limits(body_limits limits) {
404 3 : state_->body_settings_.limits = limits;
405 3 : }
406 :
407 3 : const body_limits& http_application::get_body_limits() const {
408 3 : return state_->body_settings_.limits;
409 : }
410 :
411 3 : const body_settings& http_application::get_body_settings() const {
412 3 : return state_->body_settings_;
413 : }
414 :
415 6 : void http_application::accept_body_types(std::vector<std::string> types) {
416 18 : for (auto& type : types) ::thinger::util::ascii::to_lower(type);
417 6 : state_->body_settings_.accepted = std::move(types);
418 6 : }
419 :
420 : // Request processing
421 : // The connection cannot be reused after this request (unread or broken body, failed
422 : // handler...): close it once the response is sent, for `reason`. The head of the response,
423 : // unless already written, also tells the client (see http_stream::must_close())
424 2373 : static void must_close(request& req, detail::close_reason reason) {
425 2373 : if (auto stream = req.get_http_stream()) stream->must_close(reason);
426 2373 : }
427 :
428 : // An expectation other than 100-continue cannot be met (RFC 9110 section 10.1.1): 417, as
429 : // Node and Go answer. A Host its TLS handshake (SNI) was not made for: 421, as nginx answers,
430 : // before its host runs anything (see request::misdirected()). Their body, sent or not, is
431 : // not read: the connection is not reused then. True if the request was answered so
432 : // (`head` is its http_request, which the caller holds already)
433 27340 : static bool refused_before_routing(request& req, const http_request& head, response& res) {
434 27340 : bool unmet = head.get_expectation() == http_request::expectation::unsupported;
435 27340 : if (!unmet && !req.misdirected()) [[likely]] return false;
436 84 : if (req.has_pending_body()) must_close(req, detail::close_reason::body_unread);
437 162 : if (unmet) res.error(http_response::status::expectation_failed, "Expectation Failed");
438 135 : else res.error(http_response::status::misdirected_request, "Misdirected Request");
439 84 : return true;
440 : }
441 :
442 : // The limits of the matched route apply to its request: its body size, else that of its host
443 : // (which also applies to its not-found handler), else the application's; and the reasons that
444 : // interrupt its handler (none without a route)
445 27136 : static void apply_route_limits(request& req, const virtual_host& host, const route* matched_route) {
446 27136 : const auto& limit = matched_route && matched_route->get_max_body_size() ? matched_route->get_max_body_size()
447 26824 : : host.get_max_body_size();
448 27136 : if (limit) req.set_max_body_size(*limit);
449 27136 : if (auto stream = req.get_http_stream(); stream && matched_route) {
450 25953 : stream->set_cancel_on(matched_route->get_cancel_on());
451 27136 : }
452 27136 : }
453 :
454 : // The request deadline of the matched route (or its group), else that of its host: none if
455 : // neither sets one (that of the connection applies)
456 27136 : static const std::optional<std::chrono::milliseconds>& request_deadline_of(const virtual_host& host,
457 : const route* matched_route) {
458 27136 : return matched_route && matched_route->get_request_timeout() ? matched_route->get_request_timeout()
459 27136 : : host.get_request_timeout();
460 : }
461 :
462 : // After the handler: a failed one gets a 500 unless its response was started (then the
463 : // connection is not reused, see serve()); an unreadable body fails the request, at an
464 : // unknown position
465 24864 : static void answer_failures(request& req, response& res, bool handler_failed) {
466 24966 : if (handler_failed && !res.has_responded()) res.error(http_response::status::internal_server_error);
467 24864 : if (auto error = req.get_body_error(); error != body_error::none) {
468 741 : must_close(req, detail::close_reason::body_error);
469 741 : if (!res.has_responded()) {
470 600 : if (error == body_error::too_large) {
471 879 : res.error(http_response::status::payload_too_large, "Payload Too Large");
472 307 : } else if (error == body_error::timeout) {
473 0 : res.error(http_response::status::timed_out, "Request Timeout");
474 : } else {
475 921 : res.error(http_response::status::bad_request, "Invalid request body");
476 : }
477 : }
478 : }
479 24864 : }
480 :
481 : // How serve() dispatches a request once the middlewares let it through
482 : enum class dispatch_kind { none, unmatched, handle, read_then_handle };
483 :
484 26894 : static dispatch_kind dispatch_of(request& req, response& res, const route* matched_route) {
485 : // No route matched → fallback / 404
486 26894 : if (!matched_route) return dispatch_kind::unmatched;
487 : // UNSUPPORTED TYPE: answered 415 before reading the body (dropped afterwards)
488 25753 : if (req.has_pending_body() && !matched_route->accepts_body(req, res)) return dispatch_kind::none;
489 : // DEFERRED: handler reads body at its discretion (bounded by the maximum body size in every
490 : // read); or NO BODY: dispatch directly
491 25666 : if (matched_route->is_deferred_body() || !req.has_pending_body()) return dispatch_kind::handle;
492 : // PENDING BODY: read (up to the maximum body size, or failing the request), then dispatch
493 1383 : return dispatch_kind::read_then_handle;
494 : }
495 :
496 : // What a handler failed with, for the log
497 78 : static std::string describe(const std::exception_ptr& error) {
498 : try {
499 156 : std::rethrow_exception(error);
500 78 : } catch (const std::exception& e) {
501 78 : return e.what();
502 78 : } catch (...) {
503 0 : return "unknown exception";
504 0 : }
505 : }
506 :
507 : // A failed handler may have left the response it started incomplete (unless the request timeout
508 : // answered it, `timed_out`): the connection is not reused then, closed as aborted if the scope
509 : // interrupted it (see serve())
510 26424 : static void close_if_incomplete(request& req, const response& res, const http_stream* stream, bool failed,
511 : bool timed_out) {
512 26424 : if (failed && res.has_responded() && !timed_out) {
513 15 : must_close(req, stream && stream->handler_interrupted() ? detail::close_reason::aborted
514 : : detail::close_reason::handler_failed);
515 : }
516 26424 : }
517 :
518 : // Once the handler ended (see serve()): what is logged and answered for it, and whether the body
519 : // left unread is to be discarded. `timed_out`: the request timeout answered it
520 26424 : static bool settle(request& req, response& res, const http_stream* stream,
521 : const std::optional<std::string>& failure, bool timed_out) {
522 26424 : auto http_request = req.get_http_request();
523 :
524 : // Interrupted by the cancellation of its scope (see route::cancel_on): not a failure, and
525 : // nothing more is awaited for it. No 500: the timeout already answered, the client is
526 : // gone, or the connection closes. A response it started may be incomplete
527 26424 : if (stream && stream->handler_interrupted()) [[unlikely]] {
528 4257 : LOG_DEBUG("Handler interrupted: {} {}{}", get_method(http_request->get_method()), http_request->get_path(),
529 : failure ? " (" + *failure + ")" : "");
530 1419 : if (res.has_responded() && !timed_out) must_close(req, detail::close_reason::aborted);
531 1419 : if (req.has_pending_body()) must_close(req, detail::close_reason::body_unread);
532 1419 : return false;
533 : }
534 :
535 25005 : if (failure) {
536 66 : LOG_ERROR("Exception handling {} {}: {}", get_method(http_request->get_method()),
537 : http_request->get_path(), *failure);
538 : }
539 :
540 : // Answered by the request timeout, which already decided what happens to the connection:
541 : // the handler is just late
542 25005 : if (timed_out) return false;
543 24864 : answer_failures(req, res, failure.has_value());
544 24864 : return req.has_pending_body();
545 26424 : }
546 :
547 : // An in-memory dispatch has no coroutine bound to the handler scope of its stream yet: `served`
548 : // runs on one of its own, bound to the handler signal of the stream, which is emitted from now
549 : // on. What it failed with if interrupted before its handler started, nothing once it ran
550 3960 : static awaitable<std::optional<std::string>> serve_on_own_coroutine(http_stream& stream,
551 : awaitable<void> served) {
552 : stream.bind_handler();
553 : stream.handler_started();
554 : std::optional<std::string> failure;
555 : try {
556 : co_await co_spawn(co_await boost::asio::this_coro::executor, std::move(served),
557 : boost::asio::bind_cancellation_slot(stream.handler_slot(), boost::asio::use_awaitable));
558 : co_return std::nullopt;
559 : } catch (const std::exception& e) {
560 : // Thrown once its handler ended: as if from serve()
561 : if (!stream.handler_running()) throw;
562 : // Interrupted before it started
563 : failure = e.what();
564 : }
565 : stream.handler_ended();
566 : co_return failure;
567 7920 : }
568 :
569 : namespace {
570 :
571 : // Tells the observer of a response (if any) that the application is done with its request,
572 : // however handle() ends
573 : struct handled_guard {
574 : std::shared_ptr<response_observer> observer;
575 27559 : ~handled_guard() {
576 27559 : if (observer) observer->handled();
577 27559 : }
578 : };
579 :
580 : }
581 :
582 767 : http_application::state::cors_step http_application::state::decide_cors(const http_request& head, response& res) const {
583 767 : detail::cors_decision decision;
584 767 : auto step = cors_step::go_on;
585 : try {
586 767 : decision = detail::decide_cors(*cors_, head);
587 755 : if (decision.answer_preflight) step = cors_step::answer_preflight;
588 12 : } catch (...) {
589 12 : LOG_ERROR("Exception in the CORS origin filter: {}", describe(std::current_exception()));
590 12 : step = cors_step::fail;
591 12 : }
592 767 : res.set_cors(cors_, std::move(decision));
593 767 : return step;
594 767 : }
595 :
596 120 : awaitable<void> http_application::state::answer_cors(request& req, response& res, cors_step step) {
597 : if (step == cors_step::answer_preflight) {
598 : // For every host and path, with no route whose limits apply, and no middleware (a
599 : // preflight never has credentials). Any body it has is dropped, not taken as the
600 : // next request
601 : auto no_content = std::make_shared<http_response>();
602 : no_content->set_status(http_response::status::no_content);
603 : res.send_response(no_content);
604 : if (req.has_pending_body()) co_await discard_unread_body(req);
605 : } else {
606 : // Nothing of the application runs on a request whose origin could not be decided; its
607 : // body, sent or not, is not read: the connection is not reused then
608 : if (req.has_pending_body()) must_close(req, detail::close_reason::body_unread);
609 : res.error(http_response::status::internal_server_error);
610 : }
611 240 : }
612 :
613 9 : awaitable<void> http_application::handle(std::shared_ptr<request> req, std::shared_ptr<response_sink> sink) {
614 : // Its coroutine, not one more around it
615 9 : return state_->handle(std::move(req), std::move(sink), nullptr);
616 : }
617 :
618 27559 : awaitable<void> http_application::state::handle(std::shared_ptr<request> req, std::shared_ptr<response_sink> sink,
619 : std::function<void(const response&)> created) {
620 : // Told when the response ends, while a response hook is set
621 : handled_guard handled;
622 : if (response_hook_) [[unlikely]] {
623 : handled.observer = std::make_shared<response_observer>(req, response_hook_);
624 : sink->observe(handled.observer);
625 : }
626 :
627 : response res(std::move(sink), req->get_http_request());
628 : auto http_request = req->get_http_request();
629 :
630 : // Server settings the request and its response depend on
631 : res.set_error_formatter(error_formatter_);
632 : auto cors_next = cors_ ? decide_cors(*http_request, res) : cors_step::go_on;
633 : if (created) created(res);
634 : req->set_trusted_proxies(trusted_proxies_);
635 : req->set_max_body_size(max_body_size_);
636 : req->set_body_settings(&body_settings_);
637 :
638 : // Ambiguous body framing (invalid Content-Length, unsupported Transfer-Encoding, or
639 : // both): where this request ends, and so where the next one starts, is unknown
640 : if (!http_request->has_valid_framing()) {
641 : LOG_DEBUG("Rejecting request with invalid body framing: {} {}",
642 : get_method(http_request->get_method()), http_request->get_path());
643 : must_close(*req, detail::close_reason::body_error);
644 : res.error(http_response::status::bad_request, "Invalid request framing");
645 : co_return;
646 : }
647 :
648 : if (refused_before_routing(*req, *http_request, res)) [[unlikely]] co_return;
649 :
650 : if (cors_next != cors_step::go_on) [[unlikely]] {
651 : co_await answer_cors(*req, res, cors_next);
652 : co_return;
653 : }
654 :
655 : // Match the virtual host (by the Host header), then the route among its routes, whose
656 : // limits apply from here
657 : virtual_host* host = default_host_.get();
658 : const route* matched_route = nullptr;
659 : try {
660 : host = &resolve_host(*req);
661 : matched_route = host->router().find_route(req);
662 : } catch (const std::exception& e) {
663 : LOG_ERROR("Exception matching {} {}: {}", get_method(http_request->get_method()), http_request->get_path(), e.what());
664 : must_close(*req, detail::close_reason::handler_failed);
665 : res.error(http_response::status::internal_server_error);
666 : co_return;
667 : }
668 : apply_route_limits(*req, *host, matched_route);
669 : // Measured by the connection, from the start of the request
670 : if (const auto& deadline = request_deadline_of(*host, matched_route); deadline && req->source_) {
671 : req->source_->set_request_deadline(*deadline);
672 : }
673 : bool route_timeout = matched_route && matched_route->get_timeout();
674 : const auto& timeout = route_timeout ? matched_route->get_timeout() : host->get_timeout();
675 : if (timeout && timeout->count() > 0) {
676 : auto status = route_timeout ? matched_route->get_timeout_status() : host->get_timeout_status();
677 : co_await serve_with_timeout(std::move(req), std::move(res), *host, matched_route, *timeout, status);
678 : co_return;
679 : }
680 : co_await serve(std::move(req), res, *host, matched_route);
681 55118 : }
682 :
683 27136 : bool http_application::state::start_handler(http_stream* stream) const {
684 : // Nothing starts once the application is gone. Checked in the step marking the handler
685 : // as started: close() cancels its scope later, on this executor, or this sees it closed.
686 : // Left unanswered: the request has no one to answer to (see close())
687 27136 : if (closed()) [[unlikely]] {
688 12 : if (stream) stream->handler_ended();
689 12 : return false;
690 : }
691 : // The signals of the scope are emitted only while the handler runs (on this executor)
692 27124 : if (stream) stream->handler_started();
693 27124 : return true;
694 : }
695 :
696 31096 : awaitable<void> http_application::state::serve(std::shared_ptr<request> req, response& res, virtual_host& host,
697 : const route* matched_route) {
698 : // The handler runs within the cancellation scope of its stream, on a coroutine bound to it
699 : // (the only binding of the scope, see route::cancel_on): this one, bound already (the read
700 : // loop of the connection, or the coroutine of a route timeout), with no coroutine more per
701 : // request. Without a stream there is no scope
702 : auto stream = req->get_http_stream();
703 : std::optional<std::string> failure;
704 : if (stream && !stream->handler_bound()) [[unlikely]] {
705 : failure = co_await serve_on_own_coroutine(*stream, serve(req, res, host, matched_route));
706 : if (!failure) co_return;
707 : } else {
708 : if (!start_handler(stream.get())) [[unlikely]] co_return;
709 : // Any exception is answered below with 500 unless the scope interrupted the handler
710 : try {
711 : if (!middlewares_.empty() && !co_await run_middlewares(*req, res)) {
712 : // STOPPED by a middleware, which answered (before reading the body), or the
713 : // application is gone
714 : } else if (auto how = dispatch_of(*req, res, matched_route); how == dispatch_kind::unmatched) {
715 : co_await host.router().handle_unmatched(*req, res);
716 : } else if (how == dispatch_kind::handle ||
717 : (how == dispatch_kind::read_then_handle && co_await req->read_body())) {
718 : co_await matched_route->handle_request_coro(*req, res);
719 : }
720 : } catch (...) {
721 : failure = describe(std::current_exception());
722 : }
723 :
724 : // A response it started may be incomplete: the connection closes after it, decided now,
725 : // before its head may be written (this coroutine ends after a post)
726 : close_if_incomplete(*req, res, stream.get(), failure.has_value(), res.timed_out());
727 : if (stream) {
728 : stream->handler_ended();
729 : // What the scope or the handler left in the cancellation state of this coroutine goes
730 : // with the handler (a cancellation, its own state or filter): the next request starts anew
731 : co_await boost::asio::this_coro::reset_cancellation_state();
732 : co_await boost::asio::this_coro::throw_if_cancelled(true);
733 : }
734 : }
735 :
736 : // Drop any body left unread (unmatched route, or a handler that did not read it all)
737 : // so it is not parsed as the next request
738 : if (settle(*req, res, stream.get(), failure, res.timed_out())) co_await discard_unread_body(*req);
739 62192 : }
740 :
741 : namespace {
742 :
743 : // Request timeout of a route (see route::timeout), shared by the request, its timer and
744 : // its response. Everything happens on the executor of the request.
745 : struct request_deadline {
746 261 : explicit request_deadline(const boost::asio::any_io_executor& executor)
747 261 : : timer(executor), wake(executor) {}
748 :
749 : // Expires at the deadline, unless cancelled once the response starts
750 : boost::asio::steady_timer timer;
751 : // Wakes the request once its handler ends, or it times out
752 : asio::async_event wake;
753 : bool served = false;
754 : bool expired = false;
755 : };
756 :
757 : }
758 :
759 261 : awaitable<void> http_application::state::serve_with_timeout(std::shared_ptr<request> req, response res,
760 : virtual_host& host, const route* matched_route,
761 : std::chrono::milliseconds timeout,
762 : http_response::status status) {
763 : auto executor = co_await boost::asio::this_coro::executor;
764 : auto deadline = std::make_shared<request_deadline>(executor);
765 :
766 : // The timer stops once the response starts, whatever thread starts it
767 258 : res.on_start([weak = std::weak_ptr<request_deadline>(deadline)] {
768 258 : auto state = weak.lock();
769 258 : if (!state) return;
770 516 : boost::asio::dispatch(state->timer.get_executor(), [state] { state->timer.cancel(); });
771 258 : });
772 :
773 : // The connection timeout does not close the connection while the request waits for
774 : // its response within its own timeout
775 : if (req->source_) req->source_->hold_until(std::chrono::steady_clock::now() + timeout);
776 :
777 : // On expiry, the request is answered with an error. A body still unread may be being
778 : // read by the handler, which goes on: the connection closes after the response
779 : deadline->timer.expires_after(timeout);
780 258 : deadline->timer.async_wait([deadline, req, res, status](const boost::system::error_code& ec) mutable {
781 258 : if (ec) return;
782 186 : if (!res.has_responded()) {
783 186 : bool pending_body = req->has_pending_body();
784 186 : if (pending_body) must_close(*req, detail::close_reason::route_timeout);
785 186 : if (res.answer_timeout(status)) {
786 186 : auto http_request = req->get_http_request();
787 186 : LOG_DEBUG("Request timed out: {} {}{}", get_method(http_request->get_method()),
788 : http_request->get_path(), pending_body ? " (body not read, closing the connection)" : "");
789 186 : deadline->expired = true;
790 186 : deadline->wake.set();
791 186 : }
792 : }
793 : // Then the scope of the request is cancelled, after the answer, also when the response
794 : // started (on another thread) as the timer expired: its handler goes on unless its
795 : // route declared the timeout (see route::cancel_on)
796 186 : if (auto stream = req->get_http_stream()) stream->cancel(cancel_reason::timeout);
797 : });
798 :
799 : // The handler runs on its own coroutine, so the request can end before it does: it
800 : // keeps a copy of the response, and the request (with its cancellation scope, to which
801 : // the coroutine is bound)
802 : auto stream = req->get_http_stream();
803 : if (stream) stream->bind_handler();
804 : co_spawn(executor,
805 261 : [self = shared_from_this(), req, res, &host, matched_route]() mutable -> awaitable<void> {
806 : co_await self->serve(req, res, host, matched_route);
807 522 : },
808 : // Its end wakes the request (serve() answers what the handler throws)
809 : boost::asio::bind_cancellation_slot(stream ? stream->handler_slot() : boost::asio::cancellation_slot{},
810 222 : [deadline](const std::exception_ptr&) {
811 222 : deadline->served = true;
812 222 : deadline->wake.set();
813 222 : }));
814 :
815 : // Wait until the handler ends (a response it sends later is still awaited by the
816 : // connection, and the timer is still armed), or the request times out
817 : while (!deadline->served && !deadline->expired) {
818 : if (!co_await deadline->wake.wait()) co_return;
819 : }
820 522 : }
821 :
822 : // In-memory dispatch
823 2721 : awaitable<std::shared_ptr<http_response>> http_application::dispatch(std::shared_ptr<http_request> http_request,
824 : dispatch_options options) {
825 2721 : return state::dispatch(state_, std::move(http_request), std::move(options));
826 : }
827 :
828 4083 : awaitable<std::shared_ptr<http_response>> http_application::state::dispatch(std::shared_ptr<state> self,
829 : std::shared_ptr<http_request> http_request,
830 : dispatch_options options) {
831 : if (!http_request) co_return self->make_error_response(http_response::status::bad_request, "Missing request");
832 : if (http_request->is_chunked_transfer()) {
833 : co_return self->make_error_response(http_response::status::bad_request, "Chunked requests cannot be dispatched in memory");
834 : }
835 :
836 : // The body comes from memory (see detail::memory_source): handlers read it like on a
837 : // connection (read_body(), or read()/read_some() on deferred routes)
838 : std::string body = std::move(http_request->get_body());
839 : http_request->get_body().clear();
840 : http_request->remove_headers(header::content_length);
841 : http_request->process_header(header::content_length, std::to_string(body.size()));
842 :
843 : auto executor = co_await boost::asio::this_coro::executor;
844 : auto memory = std::make_shared<memory_response>(executor);
845 :
846 : // With a response hook the dispatch also waits for its report, within the same timeout
847 : std::optional<std::chrono::steady_clock::time_point> started;
848 : if (self->response_hook_) [[unlikely]] started = std::chrono::steady_clock::now();
849 :
850 : // Run the request on its own coroutine (in a strand, where its stream, the cancellation
851 : // scope of the request, acts), and wait for the response from the start: the timeout
852 : // also covers a handler that never finishes. Handlers may also answer later, from a copy
853 : // of the response. handle() answers whatever the middlewares, body reading or handler throw.
854 : auto strand = boost::asio::make_strand(executor);
855 : auto stream = std::make_shared<http_stream>(1, *http_request, strand);
856 : auto source = std::make_unique<tracked_source>(std::move(body), std::move(options.remote_ip), self, stream, memory);
857 : // The application is gone: nothing starts, and the caller gets what cancelling it gets
858 : if (!self->track(*source)) throw boost::system::system_error(boost::asio::error::operation_aborted);
859 : auto req = std::make_shared<request>(std::move(source), stream, http_request);
860 : req->context() = std::move(options.context);
861 : // The response, once created on the strand, to answer the timeout through it
862 : struct created_response {
863 : std::mutex mutex;
864 : std::optional<response> res;
865 : };
866 : auto created = std::make_shared<created_response>();
867 : co_spawn(strand,
868 4059 : [self, req, memory, created]() -> awaitable<void> {
869 : // The application went first: the request never reached it, nor its response hook
870 : if (self->closed()) co_return;
871 4047 : co_await self->handle(req, memory, [created](const response& res) {
872 4047 : std::scoped_lock lock(created->mutex);
873 4047 : created->res.emplace(res);
874 4047 : });
875 8118 : },
876 : // What handle() itself throws is a 500, at once (unless the response is complete)
877 4053 : [self, memory](const std::exception_ptr& error) {
878 4053 : if (error) memory->send(self->make_error_response(http_response::status::internal_server_error, ""));
879 4053 : });
880 :
881 : auto result = co_await memory->wait(options.timeout);
882 : if (!result && (co_await boost::asio::this_coro::cancellation_state).cancelled() !=
883 : boost::asio::cancellation_type::none) {
884 : // The caller cancelled the dispatch: for the request, its client is gone (its handler
885 : // goes on unless its route declared it, see route::cancel_on)
886 : stream->cancel(cancel_reason::client_gone);
887 : throw boost::system::system_error(boost::asio::error::operation_aborted);
888 : }
889 : if (!result && !memory->abandoned()) {
890 : std::optional<response> res;
891 : {
892 : std::scoped_lock lock(created->mutex);
893 : res = created->res;
894 : }
895 : result = co_await answer_dispatch_timeout(self, memory, stream, std::move(res));
896 : }
897 : // The application was destroyed without a complete response (see close())
898 : if (!result) throw boost::system::system_error(boost::asio::error::operation_aborted);
899 : // The caller resumes after the hook, which may use what the caller owns
900 : if (started) [[unlikely]] {
901 : auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - *started);
902 : if (!co_await memory->wait_reported(options.timeout - elapsed)) {
903 : stream->cancel(cancel_reason::client_gone);
904 : throw boost::system::system_error(boost::asio::error::operation_aborted);
905 : }
906 : }
907 : co_return result;
908 8166 : }
909 :
910 147 : awaitable<std::shared_ptr<http_response>> http_application::state::answer_dispatch_timeout(
911 : std::shared_ptr<state> self, std::shared_ptr<memory_response> memory, std::shared_ptr<http_stream> stream,
912 : std::optional<response> res) {
913 : // Answered with the timeout as a route timeout does (a later answer is dropped, and
914 : // logged as late), without the headers the handler may still be setting on the strand;
915 : // unless an answer started first. Before the response exists, the 504 is collected alone
916 : if (res) {
917 : res->answer_timeout(http_response::status::gateway_timeout, false);
918 : } else {
919 : memory->send(self->make_error_response(http_response::status::gateway_timeout, "No response within the timeout"));
920 : }
921 : // Then the scope of the request is cancelled, on the strand: its handler goes on unless
922 : // its route declared the timeout (see route::cancel_on)
923 : stream->cancel(cancel_reason::timeout);
924 : auto result = co_await memory->wait(std::chrono::milliseconds(0));
925 : // An answer that started first and is not complete
926 : if (!result && !memory->abandoned()) {
927 : result = self->make_error_response(http_response::status::gateway_timeout, "No response within the timeout");
928 : }
929 : co_return result;
930 294 : }
931 :
932 1362 : void http_application::dispatch(const boost::asio::any_io_executor& executor,
933 : std::shared_ptr<http_request> request,
934 : sync_callback<void(std::shared_ptr<http_response>)> callback,
935 : dispatch_options options) {
936 1362 : co_spawn(executor, state::dispatch(state_, std::move(request), std::move(options)),
937 2724 : [state = state_, callback = std::move(callback)](std::exception_ptr error, std::shared_ptr<http_response> result) {
938 1356 : if (error && state->closed()) {
939 : // The application was destroyed: no error to report to a callback, so a status
940 : // saying what happened (RFC 9110 section 15.6.4)
941 1440 : result = state->make_error_response(http_response::status::service_unavailable, "Service Unavailable");
942 876 : } else if (error || !result) {
943 0 : result = state->make_error_response(http_response::status::internal_server_error, "");
944 : }
945 1356 : callback(std::move(result));
946 1356 : });
947 1362 : }
948 :
949 2130 : awaitable<bool> http_application::state::run_middlewares(request& req, response& res) {
950 : for (const auto& middleware : middlewares_) {
951 : bool next = false;
952 : try {
953 : next = co_await middleware(req, res);
954 : } catch (const std::exception& e) {
955 : LOG_ERROR("Exception in middleware: {}", e.what());
956 : if (!res.has_responded()) res.error(http_response::status::internal_server_error);
957 : co_return false;
958 : }
959 :
960 : if (res.has_responded()) co_return false;
961 :
962 : // A middleware may suspend: nothing else starts if the application went meanwhile, whose
963 : // cancellation may still be queued behind this step (left unanswered, see serve())
964 : if (closed()) [[unlikely]] co_return false;
965 :
966 : if (!next) {
967 : // stopped without responding: answer anyway so the stream is not left pending
968 : LOG_ERROR("Middleware stopped the request without sending a response");
969 : res.error(http_response::status::internal_server_error);
970 : co_return false;
971 : }
972 : }
973 : co_return true;
974 4260 : }
975 :
976 236 : awaitable<void> http_application::state::discard_unread_body(request& req) {
977 : auto stream = req.get_http_stream();
978 : // A connection taken over belongs to its new owner, unread body included
979 : if (stream && stream->takes_over()) {
980 : co_return;
981 : }
982 :
983 : // Not asked for (no 100 Continue sent), the body may never come: the connection is closed
984 : // after the response instead of waiting for it
985 : if (stream && (stream->get_continue() == http_stream::continue_state::pending ||
986 : stream->get_continue() == http_stream::continue_state::skipped)) {
987 : must_close(req, detail::close_reason::continue_skipped);
988 : co_return;
989 : }
990 :
991 : // Consume a body nobody read so the connection can be reused for the next request,
992 : // or close it after the response if the body is too large or cannot be read.
993 : if (!co_await req.discard_body(max_body_size_)) {
994 : if (stream) stream->must_close(detail::close_reason::body_unread);
995 : }
996 472 : }
997 :
998 : } // namespace thinger::http
|
