Line data Source code
1 : #include "response.hpp"
2 : #include "http_application.hpp"
3 : #include "request.hpp"
4 : #include "mime_types.hpp"
5 : #include "body_decoders.hpp"
6 : #include "response_compression.hpp"
7 : #include "file_server/file_info.hpp"
8 : #include "file_server/file_response.hpp"
9 : #include "../common/syntax.hpp"
10 : #include "../../util/ascii.hpp"
11 : #include "../../util/logger.hpp"
12 : #include "../../util/base64.hpp"
13 : #include "../../util/sha1.hpp"
14 : #include "response_sink.hpp"
15 : #include "server_connection.hpp"
16 : #include "http_stream.hpp"
17 : #include "connection_tracker.hpp"
18 : #include "detail/spawn_handler.hpp"
19 : #include "websocket_connection.hpp"
20 : #include "sse_connection.hpp"
21 : #include "../../asio/sockets/websocket.hpp"
22 : #include <algorithm>
23 : #include <atomic>
24 : #include <mutex>
25 : #include <optional>
26 :
27 : namespace thinger::http {
28 :
29 : // State shared by the copies of a response. Each answer builds its own message and sends it
30 : // only if it is the first one (responded); the others drop theirs
31 : struct response::state {
32 : std::shared_ptr<response_sink> sink;
33 : std::shared_ptr<http_request> request;
34 : std::optional<http_response> draft; // status and headers set so far
35 : std::shared_ptr<const error_formatter> formatter; // default format if null
36 : std::atomic<bool> responded{false};
37 : std::shared_ptr<const detail::cors_policy> cors; // off if null
38 : detail::cors_decision cors_decision; // what it means for this request
39 :
40 : // Request timeout: called once the response starts, and whether it was answered by it
41 : std::function<void()> on_start;
42 : std::atomic<bool> timed_out{false};
43 :
44 : // Chunked response: chunks are only written between its headers and its end
45 : enum class chunked_phase { none, open, ended };
46 : std::mutex chunked_mutex;
47 : chunked_phase chunked = chunked_phase::none;
48 :
49 : // Streamed response (under chunked_mutex too): its phase, framing, and the bytes left
50 : // of a body with Content-Length
51 : chunked_phase stream = chunked_phase::none;
52 : body_framing stream_framing = body_framing::none;
53 : size_t stream_remaining = 0;
54 : };
55 :
56 24 : response::response(const std::shared_ptr<server_connection>& connection,
57 : const std::shared_ptr<http_stream>& stream,
58 24 : const std::shared_ptr<http::http_request>& http_request)
59 24 : : response(std::make_shared<connection_sink>(connection, stream), http_request) {}
60 :
61 28505 : response::response(std::shared_ptr<response_sink> sink,
62 28505 : const std::shared_ptr<http::http_request>& http_request)
63 28505 : : state_(std::make_shared<state>()) {
64 28505 : state_->sink = std::move(sink);
65 28505 : state_->request = http_request;
66 28505 : }
67 :
68 767 : void response::set_cors(std::shared_ptr<const detail::cors_policy> policy, detail::cors_decision decision) {
69 767 : state_->cors = std::move(policy);
70 767 : state_->cors_decision = std::move(decision);
71 767 : }
72 :
73 3864 : void response::log_late_answer(const char* message) const {
74 3864 : if (state_->timed_out) {
75 222 : LOG_DEBUG("{} (the request timed out)", message);
76 : } else {
77 3642 : LOG_ERROR("{}", message);
78 : }
79 3864 : }
80 :
81 34668 : bool response::ensure_not_responded() const {
82 34668 : if (state_->responded) {
83 3832 : log_late_answer("Response already sent");
84 3832 : return false;
85 : }
86 30836 : return true;
87 : }
88 :
89 27549 : bool response::mark_responded() {
90 27549 : if (state_->responded.exchange(true)) {
91 14 : log_late_answer("Response already sent");
92 14 : return false;
93 : }
94 27535 : if (state_->on_start) state_->on_start();
95 27535 : return true;
96 : }
97 :
98 261 : void response::on_start(std::function<void()> hook) {
99 261 : state_->on_start = std::move(hook);
100 261 : }
101 :
102 333 : bool response::answer_timeout(http::http_response::status status, bool with_headers) {
103 333 : if (state_->responded) return false;
104 : // Set before answering: whoever sees the response sent sees why
105 333 : state_->timed_out = true;
106 333 : if (state_->responded.exchange(true)) {
107 : // Another answer started meanwhile: it was not late
108 0 : state_->timed_out = false;
109 0 : return false;
110 : }
111 333 : if (state_->on_start) state_->on_start();
112 333 : auto message = take_draft(with_headers);
113 333 : message->set_status(status);
114 666 : format_error(state_->formatter.get(), {status, "No response within the timeout", nullptr}, *message);
115 333 : prepare_body(*message, false);
116 333 : state_->sink->send(std::move(message));
117 333 : return true;
118 1332 : }
119 :
120 52848 : bool response::timed_out() const {
121 : // A handler may have moved the response away
122 52848 : return state_ && state_->timed_out;
123 : }
124 :
125 27265 : std::shared_ptr<http_response> response::new_response() {
126 27265 : if (!mark_responded()) return nullptr;
127 27251 : return take_draft();
128 : }
129 :
130 27584 : std::shared_ptr<http_response> response::take_draft(bool with_draft) {
131 : // Moved, not copied: only the answer sent reads it, once it is the first one
132 27584 : bool drafted = with_draft && state_->draft;
133 27584 : auto message = drafted ? std::make_shared<http_response>(std::move(*state_->draft))
134 27584 : : std::make_shared<http_response>();
135 : // Once the headers set so far are there, which it does not replace
136 27584 : if (state_->cors) detail::apply_cors(*state_->cors, state_->cors_decision, *message);
137 27584 : return message;
138 0 : }
139 :
140 3288 : http_response& response::draft() {
141 3288 : if (!state_->draft) state_->draft.emplace();
142 3288 : return *state_->draft;
143 : }
144 :
145 26038 : bool response::head_request() const {
146 26038 : return state_->request->get_method() == method::HEAD;
147 : }
148 :
149 : namespace {
150 :
151 : // Whether a message sent whole is framed as response::frame_message() leaves it: with a
152 : // Content-Length (set with its content) and no Transfer-Encoding, and a status with body
153 21764 : bool framed(const http_response& message) {
154 21764 : int code = message.get_status_code();
155 43177 : return message.has_content_length_header() && !message.has_transfer_encoding_header() &&
156 43177 : code >= 200 && code != 204 && code != 304;
157 : }
158 :
159 : }
160 :
161 387 : void response::frame_message(http_response& message) const {
162 : // As Node does, a message sent whole tells where its body ends, else the client would
163 : // read it until the connection closes (RFC 9112, 6.3): its Content-Length, never
164 : // Transfer-Encoding (its body is not chunked). 1xx, 204 and 304 never have a body, nor
165 : // its length for 1xx and 204 (RFC 9110, 8.6); a 304 may tell the length of the
166 : // representation. A response to HEAD gets the length of the content it was given, if
167 : // any: it has no body anyway.
168 387 : int code = message.get_status_code();
169 387 : if (code < 200 || code == 204 || code == 304) {
170 241 : message.get_content().clear();
171 241 : if (message.has_transfer_encoding_header()) message.remove_headers(header::transfer_encoding);
172 241 : if (code != 304 && message.has_content_length_header()) message.remove_headers(header::content_length);
173 241 : return;
174 : }
175 146 : if (message.has_transfer_encoding_header()) message.remove_headers(header::transfer_encoding);
176 146 : if (!message.has_content_length_header() && (!head_request() || !message.get_content().empty())) {
177 137 : message.set_content_length(message.get_content().size());
178 : }
179 : }
180 :
181 4317 : body_framing response::frame_takeover(http_response& message) const {
182 : // What follows the response belongs to the handler. Switching protocols (101, or 2xx to
183 : // CONNECT: a tunnel), nothing tells a body (RFC 9110, 9.3.6). Otherwise, unless the
184 : // handler framed it, what the handler writes is the body, ending when the connection
185 : // closes: the response says so, as Node does for a body of unknown length not chunked
186 4317 : int code = message.get_status_code();
187 4317 : bool tunnel = code == 101 || (state_->request->get_method() == method::CONNECT && code >= 200 && code < 300);
188 4317 : if (tunnel) {
189 3670 : message.remove_headers(header::content_length);
190 3670 : message.remove_headers(header::transfer_encoding);
191 647 : } else if (!message.has_content_length_header() && !message.has_transfer_encoding_header()) {
192 638 : return body_framing::close;
193 : }
194 3679 : return body_framing::none;
195 : }
196 :
197 21764 : void response::prepare_body(http_response& message, bool compress) const {
198 : // The usual answer is framed already: its Content-Length was set with its content
199 21764 : if (!framed(message)) [[unlikely]] {
200 387 : frame_message(message);
201 : }
202 21764 : if (head_request()) {
203 : // The headers of the answer to GET, Content-Length included, without its body: the
204 : // client reads none (RFC 9110, 9.3.2), so it would be taken as the next response.
205 : // Not compressed (like Express): its length is the one of the identity body.
206 304 : if (compress && compression::compressible(message)) compression::add_vary(message, "Accept-Encoding");
207 304 : message.get_content().clear();
208 21460 : } else if (compress && message.get_content().size() >= compression::threshold) {
209 1309 : compression::compress(message, *state_->request);
210 : }
211 21764 : }
212 :
213 21431 : void response::send_message(std::shared_ptr<http_response> message, bool compress) {
214 21431 : prepare_body(*message, compress);
215 21431 : state_->sink->send(std::move(message));
216 21431 : }
217 :
218 4509 : bool response::reject_takeover(const char* feature) {
219 4509 : if (state_->sink->can_take_over()) return false;
220 24 : error(http::http_response::status::not_implemented,
221 36 : std::string(feature) + " is not available for in-memory requests");
222 12 : return true;
223 : }
224 :
225 2785 : void response::json(const nlohmann::json& data, http::http_response::status status) {
226 2785 : if (!ensure_not_responded()) return;
227 : // Before taking the draft: dump() throws for strings that are not UTF-8, and the error
228 : // answered then keeps the headers set
229 1876 : auto body = data.dump();
230 1876 : auto message = new_response();
231 1876 : if (!message) return;
232 1876 : message->set_status(status);
233 3752 : message->set_content(std::move(body), "application/json");
234 1876 : send_message(std::move(message));
235 1876 : }
236 :
237 27 : void response::encode(const nlohmann::json& data, http::http_response::status status) {
238 27 : if (!ensure_not_responded()) return;
239 : // Repeated Accept headers make one list
240 27 : std::string accept;
241 27 : if (state_->request) {
242 42 : for (const auto& value : state_->request->get_headers_with_key(header::accept)) {
243 15 : if (!accept.empty()) accept += ", ";
244 15 : accept += value;
245 27 : }
246 : }
247 27 : auto encoding = negotiate_response_encoding(accept);
248 : // Before taking the draft, as json()
249 27 : auto body = encode_body(data, encoding.format);
250 27 : auto message = new_response();
251 27 : if (!message) return;
252 27 : message->set_status(status);
253 27 : message->set_content(std::move(body), encoding.media_type);
254 27 : compression::add_vary(*message, "Accept");
255 27 : send_message(std::move(message));
256 27 : }
257 :
258 17102 : void response::send(const std::string& text, const std::string& content_type) {
259 17103 : if (!ensure_not_responded()) return;
260 15979 : auto message = new_response();
261 15979 : if (!message) return;
262 15978 : message->set_content(text, content_type);
263 15978 : send_message(std::move(message));
264 15979 : }
265 :
266 3379 : void response::error(http::http_response::status status, const std::string& message,
267 : const nlohmann::json& details) {
268 3380 : if (!ensure_not_responded()) return;
269 2483 : auto response = new_response();
270 2483 : if (!response) return;
271 2482 : response->set_status(status);
272 4964 : format_error(state_->formatter.get(), {status, message, details}, *response);
273 2482 : send_message(std::move(response));
274 4965 : }
275 :
276 27559 : void response::set_error_formatter(std::shared_ptr<const error_formatter> formatter) {
277 27559 : state_->formatter = std::move(formatter);
278 27559 : }
279 :
280 315 : void response::status(http::http_response::status s) {
281 315 : if (!ensure_not_responded()) return;
282 315 : draft().set_status(s);
283 : }
284 :
285 2815 : void response::header(std::string key, std::string value) {
286 2815 : if (!ensure_not_responded()) return;
287 2815 : draft().add_header(std::move(key), std::move(value));
288 : }
289 :
290 57 : void response::set_header(std::string key, std::string value) {
291 57 : if (!ensure_not_responded()) return;
292 57 : draft().replace_headers(std::move(key), std::move(value));
293 : }
294 :
295 : namespace {
296 :
297 : // The headers set before a message built by hand: those of the message win over the ones
298 : // of the same name, as the headers given to writeHead() in Node. Compared only with the
299 : // fields it came with, so repeated ones (Set-Cookie) are all added. Its framing is its own
300 15 : void add_draft_headers(http_response& draft, http_response& message) {
301 15 : const auto own = static_cast<std::ptrdiff_t>(message.get_headers().size());
302 36 : for (auto& field : draft.get_headers()) {
303 39 : if (headers::is_header(field.first, header::content_length) ||
304 21 : headers::is_header(field.first, header::transfer_encoding)) continue;
305 18 : const auto& fields = message.get_headers();
306 18 : bool replaced = std::any_of(fields.begin(), fields.begin() + own, [&field](const auto& other) {
307 21 : return headers::is_header(other.first, field.first);
308 : });
309 18 : if (!replaced) message.add_header(std::move(field.first), std::move(field.second));
310 : }
311 15 : }
312 :
313 : }
314 :
315 284 : void response::send_response(const std::shared_ptr<http_response>& response) {
316 284 : if (!ensure_not_responded()) return;
317 : // A message built by hand may carry the framing of another one (e.g. copied from a
318 : // response whose body was decoded): its content is its body, sent whole (a response
319 : // to HEAD keeps the length it tells)
320 284 : int code = response->get_status_code();
321 284 : if (!head_request() && code != 304 && response->has_content_length_header()) {
322 83 : response->remove_headers(header::content_length);
323 : }
324 284 : if (!mark_responded()) return;
325 : // Before the CORS headers, which do not replace the ones set so far (see take_draft)
326 284 : if (state_->draft) add_draft_headers(*state_->draft, *response);
327 284 : if (state_->cors) detail::apply_cors(*state_->cors, state_->cors_decision, *response);
328 284 : send_message(response);
329 : }
330 :
331 647 : const http_request& response::request_message() const {
332 647 : return *state_->request;
333 : }
334 :
335 7 : const error_formatter* response::formatter() const {
336 7 : return state_->formatter.get();
337 : }
338 :
339 6382 : bool response::has_responded() const {
340 6382 : return state_->responded;
341 : }
342 :
343 282 : std::shared_ptr<server_connection> response::get_connection() const {
344 282 : return state_->sink->connection();
345 : }
346 :
347 1066 : void response::redirect(const std::string& url, http::http_response::status redirect_type) {
348 1077 : if (!ensure_not_responded()) return;
349 186 : auto response = new_response();
350 186 : if (!response) return;
351 175 : response->set_status(redirect_type);
352 : // The bytes a field value cannot hold are percent-encoded, so the redirect is still sent
353 : // (as Express's res.location); spaces, '%' and non-ASCII bytes are kept as they are
354 175 : if (syntax::is_field_value(url)) {
355 169 : response->add_header(header::location, url);
356 : } else {
357 : static constexpr char hex[] = "0123456789ABCDEF";
358 6 : std::string location;
359 6 : location.reserve(url.size() + 8);
360 108 : for (unsigned char c : url) {
361 102 : if (syntax::is_field_value_char(c)) {
362 78 : location += static_cast<char>(c);
363 : } else {
364 24 : location += '%';
365 24 : location += hex[c >> 4];
366 24 : location += hex[c & 0x0F];
367 : }
368 : }
369 6 : response->add_header(header::location, std::move(location));
370 6 : }
371 : // Without it, a keep-alive client would read the body until the connection closes
372 175 : response->set_content_length(0);
373 175 : send_message(std::move(response));
374 186 : }
375 :
376 520 : void response::send_file(const std::filesystem::path& path, bool force_download) {
377 524 : if (!ensure_not_responded()) return;
378 :
379 : // One stat for the checks, the length and the validators
380 517 : auto file = file_info::of(path);
381 517 : if (!file) {
382 6 : error(http_response::status::not_found, "File not found");
383 2 : return;
384 : }
385 515 : if (!file->regular) {
386 6 : error(http_response::status::forbidden, "Not a regular file");
387 2 : return;
388 : }
389 513 : file_server::send(*this, {path, *file, mime_types::extension_to_type(path.extension().string()), {}, force_download});
390 517 : }
391 :
392 : namespace {
393 :
394 : using ::thinger::util::ascii::iequals;
395 :
396 : // Why a WebSocket handshake is rejected: the error answered, and whether the client is
397 : // told the version supported
398 : struct handshake_rejection {
399 : http_response::status status;
400 : const char* message;
401 : bool unsupported_version = false;
402 : };
403 :
404 : // Checks of a WebSocket handshake request (RFC 6455, 4.2.1) before its subprotocol
405 3714 : std::optional<handshake_rejection> check_handshake(const http_request& request) {
406 : using status = http_response::status;
407 3714 : if (!iequals(request.get_header(header::upgrade), "websocket")) {
408 18 : return handshake_rejection{status::upgrade_required, "This service requires use of WebSockets"};
409 : }
410 3696 : if (request.get_method() != method::GET) {
411 18 : return handshake_rejection{status::bad_request, "A WebSocket handshake must be a GET request"};
412 : }
413 3678 : bool upgrade = false;
414 7356 : for (const auto& value : request.get_headers_with_key(header::connection)) {
415 7410 : upgrade = upgrade || syntax::any_element(value, [](std::string_view option) { return iequals(option, "upgrade"); });
416 3678 : }
417 3678 : if (!upgrade) return handshake_rejection{status::bad_request, "A WebSocket handshake requires Connection: Upgrade"};
418 3669 : if (request.get_header("Sec-WebSocket-Version") != "13") {
419 18 : return handshake_rejection{status::upgrade_required, "Unsupported WebSocket version", true};
420 : }
421 3651 : return std::nullopt;
422 : }
423 :
424 : // Subprotocols offered by the client, in its order of preference: the comma-separated
425 : // values of its Sec-WebSocket-Protocol headers
426 3651 : std::vector<std::string> offered_protocols(const http_request& request) {
427 3651 : std::vector<std::string> offered;
428 3944 : for (const auto& value : request.get_headers_with_key("Sec-WebSocket-Protocol")) {
429 730 : syntax::for_each_element(value, [&](std::string_view protocol) { offered.emplace_back(protocol); });
430 3651 : }
431 3651 : return offered;
432 0 : }
433 :
434 : // The first subprotocol offered that is supported, empty if none
435 3651 : std::string select_protocol(const std::vector<std::string>& offered, const std::vector<std::string>& supported) {
436 3813 : for (const auto& candidate : offered) {
437 374 : if (std::find(supported.begin(), supported.end(), candidate) != supported.end()) return candidate;
438 : }
439 3439 : return {};
440 : }
441 :
442 : // A Sec-WebSocket-Key: 16 bytes in base64 (RFC 6455, 4.1)
443 3570 : bool valid_websocket_key(const std::string& key) {
444 3570 : if (key.size() != 24 || !key.ends_with("==")) return false;
445 81489 : for (size_t i = 0; i < 22; ++i) {
446 77946 : char c = key[i];
447 77946 : if (!std::isalnum(static_cast<unsigned char>(c)) && c != '+' && c != '/') return false;
448 : }
449 3543 : return ::thinger::util::base64::decode(key).size() == 16;
450 : }
451 :
452 3543 : std::string websocket_accept(const std::string& key) {
453 3543 : return ::thinger::util::base64::encode(::thinger::util::sha1::hash(key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"));
454 : }
455 :
456 : }
457 :
458 3714 : std::shared_ptr<http_response> response::websocket_handshake(const std::vector<std::string>& protocols,
459 : bool protocol_required, bool reject_unknown_protocol,
460 : std::string& protocol) {
461 3714 : const auto& request = *state_->request;
462 3714 : if (auto rejection = check_handshake(request)) {
463 135 : if (rejection->unsupported_version) header("Sec-WebSocket-Version", "13");
464 189 : error(rejection->status, rejection->message);
465 63 : return nullptr;
466 : }
467 :
468 3651 : auto offered = offered_protocols(request);
469 3651 : protocol = select_protocol(offered, protocols);
470 3651 : bool required = protocol_required && !protocols.empty();
471 3651 : if (protocol.empty() && !offered.empty() && (reject_unknown_protocol || required)) {
472 54 : LOG_DEBUG("Unsupported WebSocket protocol: {}", request.get_header("Sec-WebSocket-Protocol"));
473 162 : error(http_response::status::bad_request, "Unsupported WebSocket protocol");
474 54 : return nullptr;
475 : }
476 3597 : if (offered.empty() && required) {
477 81 : error(http_response::status::bad_request, "This method requires specifying a WebSocket protocol");
478 27 : return nullptr;
479 : }
480 :
481 3570 : const auto& key = request.get_header("Sec-WebSocket-Key");
482 3570 : if (!valid_websocket_key(key)) {
483 81 : error(http_response::status::bad_request, "Missing or invalid Sec-WebSocket-Key header");
484 27 : return nullptr;
485 : }
486 :
487 3543 : auto response = new_response();
488 3543 : if (!response) return nullptr;
489 3543 : response->set_status(http_response::status::switching_protocols);
490 7086 : response->add_header(header::upgrade, "websocket");
491 7086 : response->add_header(header::connection, "Upgrade");
492 10629 : response->add_header("Sec-WebSocket-Accept", websocket_accept(key));
493 3967 : if (!protocol.empty()) response->add_header("Sec-WebSocket-Protocol", protocol);
494 3543 : return response;
495 3651 : }
496 :
497 0 : void detail::log_handler_exception(const char* what, const std::exception_ptr& error) {
498 : try {
499 0 : std::rethrow_exception(error);
500 0 : } catch (const std::exception& e) {
501 0 : LOG_ERROR("Exception in the {} handler: {}", what, e.what());
502 0 : } catch (...) {
503 0 : LOG_ERROR("Unknown exception in the {} handler", what);
504 0 : }
505 0 : }
506 :
507 : using detail::spawn_handler;
508 :
509 3150 : void response::upgrade_websocket(const websocket_accept_options& options, websocket_handler handler) {
510 3150 : accept_websocket(options, [handler = std::move(handler)](std::shared_ptr<websocket_connection> connection,
511 : asio::websocket&) {
512 3063 : handler(connection);
513 3063 : connection->start();
514 3063 : });
515 3150 : }
516 :
517 15 : void response::upgrade_websocket(const websocket_accept_options& options, websocket_coroutine handler) {
518 15 : accept_websocket(options, [handler = std::move(handler)](std::shared_ptr<websocket_connection> connection,
519 : asio::websocket& websocket) mutable {
520 15 : spawn_handler(websocket.get_executor(), "WebSocket",
521 45 : [handler = std::move(handler), connection]() mutable -> awaitable<void> {
522 : co_await handler(connection);
523 : connection->start();
524 30 : });
525 15 : });
526 15 : }
527 :
528 3165 : void response::accept_websocket(const websocket_accept_options& options,
529 : std::function<void(std::shared_ptr<websocket_connection>, asio::websocket&)> deliver) {
530 3252 : if (!ensure_not_responded()) return;
531 3165 : if (reject_takeover("WebSocket upgrade")) return;
532 :
533 3159 : std::string protocol;
534 3159 : auto response = websocket_handshake(options.protocols, options.protocol_required, true, protocol);
535 3159 : if (!response) return;
536 :
537 3078 : hand_over(std::move(response), [deliver = std::move(deliver), options, protocol](upgraded_connection connection) {
538 : // binary, server; on the io_context of the connection, where it is taken over (the
539 : // default executor). The connection holds the scope: told of a shutdown, counted as a
540 : // websocket once started
541 3078 : auto websocket = std::make_shared<asio::websocket>(std::move(connection.socket), true, true);
542 3078 : websocket->set_protocol(protocol);
543 : // Frames the client sent right after its request, without waiting for the response:
544 : // read first, once the connection starts reading
545 3078 : if (!connection.buffered.empty()) websocket->set_read_ahead(std::move(connection.buffered));
546 3078 : connection.scope.set_type(connection_type::websocket);
547 3078 : auto ws_connection = std::make_shared<websocket_connection>(websocket, options, std::move(connection.scope));
548 :
549 : // The handler sets its callbacks, then the WebSocket starts
550 3078 : deliver(ws_connection, *websocket);
551 3078 : });
552 3240 : }
553 :
554 21 : void response::upgrade_websocket_stream(const websocket_accept_options& options, websocket_stream_coroutine handler) {
555 21 : upgrade_websocket_stream(options, websocket_stream_handler(
556 42 : [handler = std::move(handler)](upgraded_websocket websocket) mutable {
557 21 : auto executor = websocket.stream->get_executor();
558 21 : spawn_handler(std::move(executor), "WebSocket",
559 42 : [handler = std::move(handler), websocket = std::move(websocket)]() mutable {
560 21 : return handler(std::move(websocket));
561 : });
562 21 : }));
563 21 : }
564 :
565 555 : void response::upgrade_websocket_stream(const websocket_accept_options& options, websocket_stream_handler handler) {
566 645 : if (!ensure_not_responded()) return;
567 555 : if (reject_takeover("WebSocket upgrade")) return;
568 :
569 555 : std::string protocol;
570 555 : auto response = websocket_handshake(options.protocols, options.protocol_required, false, protocol);
571 555 : if (!response) return;
572 :
573 465 : hand_over(std::move(response), [handler = std::move(handler), options, protocol](upgraded_connection connection) {
574 : // On the io_context of the connection, where it is taken over (the default executor)
575 465 : auto websocket = std::make_shared<asio::websocket>(std::move(connection.socket), options.binary, true);
576 465 : websocket->set_protocol(protocol);
577 : // Frames the client sent right after its request, without waiting for the response
578 465 : if (!connection.buffered.empty()) websocket->set_read_ahead(std::move(connection.buffered));
579 : // No keepalive unless asked for
580 930 : websocket->set_keepalive(options.ping_interval.value_or(std::chrono::milliseconds{0}),
581 465 : options.idle_timeout.value_or(std::chrono::milliseconds{0}));
582 465 : connection.scope.set_type(connection_type::websocket);
583 465 : handler(upgraded_websocket{std::move(websocket), protocol, std::move(connection.scope)});
584 465 : });
585 645 : }
586 :
587 261 : void response::take_over(takeover_handler handler) {
588 261 : if (!ensure_not_responded()) return;
589 261 : if (reject_takeover("Connection takeover")) return;
590 258 : if (auto message = new_response()) hand_over(std::move(message), std::move(handler));
591 : }
592 :
593 3 : void response::take_over(takeover_coroutine handler) {
594 3 : take_over(takeover_handler([handler = std::move(handler)](upgraded_connection connection) mutable {
595 3 : auto& io_context = connection.socket->get_io_context();
596 3 : spawn_handler(io_context, "takeover",
597 3 : [handler = std::move(handler), connection = std::move(connection)]() mutable {
598 3 : return handler(std::move(connection));
599 : });
600 3 : }));
601 3 : }
602 :
603 4317 : void response::hand_over(std::shared_ptr<http_response> message, takeover_handler handler) {
604 4317 : auto framing = frame_takeover(*message);
605 4317 : state_->sink->take_over(std::move(message), std::move(handler), framing);
606 4317 : }
607 :
608 510 : void response::start_sse(sse_handler handler, const sse_options& options) {
609 510 : accept_sse(options, [handler = std::move(handler)](std::shared_ptr<sse_connection> connection, asio::socket&) {
610 498 : handler(std::move(connection));
611 498 : });
612 510 : }
613 :
614 18 : void response::start_sse(sse_coroutine handler, const sse_options& options) {
615 18 : accept_sse(options, [handler = std::move(handler)](std::shared_ptr<sse_connection> connection, asio::socket& socket) mutable {
616 18 : spawn_handler(socket.get_io_context(), "SSE", [handler = std::move(handler), connection = std::move(connection)]() mutable {
617 18 : return handler(std::move(connection));
618 : });
619 18 : });
620 18 : }
621 :
622 528 : void response::accept_sse(const sse_options& options,
623 : std::function<void(std::shared_ptr<sse_connection>, asio::socket&)> deliver) {
624 540 : if (!ensure_not_responded()) return;
625 528 : if (reject_takeover("Server-Sent Events")) return;
626 :
627 525 : auto response = new_response();
628 525 : if (!response) return;
629 525 : response->set_status(http_response::status::ok);
630 1050 : response->set_content_type("text/event-stream");
631 2100 : response->add_header("Cache-Control", "no-cache");
632 2100 : response->add_header("X-Accel-Buffering", "no"); // Disable nginx buffering
633 :
634 : // HEAD: the headers of the event stream alone, as a whole response (no events, nor
635 : // handler), so the connection goes on with the next request
636 525 : if (head_request()) {
637 9 : send_message(std::move(response), false);
638 9 : return;
639 : }
640 :
641 : // Send the SSE headers and take over the connection (anything the client sends
642 : // afterwards is not part of the event stream)
643 516 : hand_over(std::move(response), [deliver = std::move(deliver), keepalive = options.keepalive](upgraded_connection connection) {
644 : // The connection holds the scope: told of a shutdown, counted as an event stream
645 516 : auto& socket = *connection.socket;
646 1032 : auto sse_conn = std::make_shared<sse_connection>(std::move(connection.socket), boost::asio::any_io_executor{},
647 1032 : std::move(connection.scope));
648 516 : sse_conn->set_keepalive(keepalive);
649 516 : sse_conn->start();
650 516 : deliver(std::move(sse_conn), socket);
651 516 : });
652 525 : }
653 :
654 : // Chunked response support: the sink writes the headers, then each chunk. Chunks are
655 : // written in order (the lock), only once the headers are queued and until the end. A
656 : // response to HEAD is the headers alone, as a whole response: its chunks are dropped
657 1090 : bool response::start_chunked(const std::string& content_type, http::http_response::status status) {
658 1090 : if (!ensure_not_responded()) return false;
659 1069 : auto response = new_response();
660 1069 : if (!response) return false;
661 1068 : response->set_status(status);
662 1068 : response->set_content_type(content_type);
663 :
664 1068 : std::lock_guard lock(state_->chunked_mutex);
665 1068 : if (head_request()) {
666 : // Without Transfer-Encoding, as start_stream() does
667 9 : state_->sink->send(std::move(response));
668 1059 : } else if (!state_->sink->begin(std::move(response))) {
669 0 : return false;
670 : }
671 1068 : state_->chunked = state::chunked_phase::open;
672 1068 : return true;
673 1069 : }
674 :
675 1198 : bool response::write_chunk(const std::string& data) {
676 1198 : std::lock_guard lock(state_->chunked_mutex);
677 1198 : if (state_->chunked != state::chunked_phase::open) {
678 9 : log_late_answer("Chunks can only be written between start_chunked() and end_chunked()");
679 9 : return false;
680 : }
681 1189 : if (head_request()) return true;
682 1171 : return state_->sink->append(data);
683 1198 : }
684 :
685 1071 : bool response::end_chunked() {
686 1071 : std::lock_guard lock(state_->chunked_mutex);
687 1071 : if (state_->chunked != state::chunked_phase::open) {
688 9 : log_late_answer("Only a response started with start_chunked() can be ended, once");
689 9 : return false;
690 : }
691 1062 : state_->chunked = state::chunked_phase::ended;
692 1062 : if (head_request()) return true;
693 1053 : return state_->sink->finish();
694 1071 : }
695 :
696 :
697 : namespace {
698 :
699 1722 : bool is_framing_header(std::string_view name) {
700 3432 : return headers::is_header(name, header::content_length) || headers::is_header(name, header::transfer_encoding) ||
701 3432 : headers::is_header(name, header::connection);
702 : }
703 :
704 : // Status and headers of the head of a streamed response: those of `head` replace the ones
705 : // with the same name set before (e.g. CORS), except Vary, which adds up: the response
706 : // depends on the fields of both, and a cache must know. The framing is the response's own
707 719 : void apply_head(http_response& message, const http_response& head) {
708 719 : message.set_status(head.get_status());
709 719 : message.set_reason_phrase(head.get_reason());
710 1580 : for (const auto& [key, value] : head.get_headers()) {
711 861 : if (!is_framing_header(key) && !headers::is_header(key, "Vary")) message.remove_headers(key);
712 : }
713 : // Credentials granted by CORS mean nothing next to an origin "*", and browsers reject them
714 719 : if (head.get_header("Access-Control-Allow-Origin") == "*") message.remove_headers("Access-Control-Allow-Credentials");
715 719 : const bool merge_vary = message.has_header("Vary");
716 1580 : for (const auto& [key, value] : head.get_headers()) {
717 861 : if (is_framing_header(key)) continue;
718 855 : if (merge_vary && headers::is_header(key, "Vary")) {
719 108 : syntax::for_each_element(value, [&message](std::string_view field) { compression::add_vary(message, field); });
720 : } else {
721 804 : message.add_header(key, value);
722 : }
723 : }
724 : // headers set before (header()) may carry framing too
725 719 : message.remove_headers(header::content_length);
726 719 : message.remove_headers(header::transfer_encoding);
727 719 : }
728 :
729 240 : bool is_http_1_1(const http_request& request) {
730 480 : return request.get_http_version_major() > 1 ||
731 480 : (request.get_http_version_major() == 1 && request.get_http_version_minor() >= 1);
732 : }
733 :
734 : // Framing of a streamed response, with its Content-Length if it has one
735 719 : body_framing frame_stream(http_response& message, const http_request& request,
736 : std::optional<size_t> content_length) {
737 719 : int code = message.get_status_code();
738 719 : bool no_body = request.get_method() == method::HEAD || (code >= 100 && code < 200) || code == 204 || code == 304;
739 719 : if (no_body) {
740 : // the length of the resource (HEAD, 304), but never for 1xx and 204
741 18 : if (content_length && code >= 200 && code != 204) message.set_content_length(*content_length);
742 18 : return body_framing::none;
743 : }
744 701 : if (content_length) {
745 461 : message.set_content_length(*content_length);
746 461 : return body_framing::length;
747 : }
748 240 : return is_http_1_1(request) ? body_framing::chunked : body_framing::close;
749 : }
750 :
751 : }
752 :
753 : // Streamed response: the sink writes the head, then each part of the body once the previous
754 : // one is written. The response checks the body against its framing, so a writer can never
755 : // put more on the connection than the head announced (it would be read as what follows)
756 719 : bool response::start_stream(const http_response& head, std::optional<size_t> content_length) {
757 719 : if (!ensure_not_responded()) return false;
758 719 : auto message = new_response();
759 719 : if (!message) return false;
760 719 : apply_head(*message, head);
761 719 : auto framing = frame_stream(*message, *state_->request, content_length);
762 :
763 719 : std::lock_guard lock(state_->chunked_mutex);
764 719 : if (!state_->sink->begin_stream(std::move(message), framing)) return false;
765 719 : state_->stream = state::chunked_phase::open;
766 719 : state_->stream_framing = framing;
767 719 : state_->stream_remaining = framing == body_framing::length ? *content_length : 0;
768 719 : return true;
769 719 : }
770 :
771 14594 : awaitable<bool> response::write_stream(std::string data) {
772 : {
773 : std::lock_guard lock(state_->chunked_mutex);
774 : if (state_->stream != state::chunked_phase::open) {
775 : log_late_answer("Stream data can only be written between start_stream() and end_stream()");
776 : co_return false;
777 : }
778 : if (state_->stream_framing == body_framing::none) {
779 : if (!data.empty()) LOG_ERROR("Response without body: stream data dropped");
780 : co_return data.empty();
781 : }
782 : if (state_->stream_framing == body_framing::length) {
783 : if (data.size() > state_->stream_remaining) {
784 : LOG_ERROR("Stream data exceeds the Content-Length of the response: dropped");
785 : co_return false;
786 : }
787 : state_->stream_remaining -= data.size();
788 : }
789 : }
790 : // An empty chunk would end a chunked body
791 : if (data.empty()) co_return true;
792 : co_return co_await state_->sink->write_stream(std::move(data));
793 29188 : }
794 :
795 675 : awaitable<bool> response::end_stream() {
796 : {
797 : std::lock_guard lock(state_->chunked_mutex);
798 : if (state_->stream != state::chunked_phase::open) {
799 : log_late_answer("Only a response started with start_stream() can be ended, once");
800 : co_return false;
801 : }
802 : state_->stream = state::chunked_phase::ended;
803 : if (state_->stream_framing == body_framing::length && state_->stream_remaining > 0) {
804 : LOG_ERROR("Stream ended {} bytes before its Content-Length: closing the connection",
805 : state_->stream_remaining);
806 : state_->sink->abort_stream();
807 : co_return false;
808 : }
809 : }
810 : co_return co_await state_->sink->end_stream();
811 1350 : }
812 :
813 61 : void response::abort_stream() {
814 61 : std::lock_guard lock(state_->chunked_mutex);
815 61 : if (state_->stream != state::chunked_phase::open) return;
816 43 : state_->stream = state::chunked_phase::ended;
817 43 : state_->sink->abort_stream();
818 61 : }
819 :
820 : } // namespace thinger::http
821 :
|
