Line data Source code
1 : #include "client_connection.hpp"
2 : #include "client_error.hpp"
3 : #include "../../asio/sockets/ssl_socket.hpp"
4 : #include "../../util/ascii.hpp"
5 : #include "../../util/logger.hpp"
6 : #include "../../util/compression.hpp"
7 : #include <charconv>
8 : #include <boost/asio/ssl/error.hpp>
9 :
10 : namespace thinger::http {
11 :
12 : namespace {
13 :
14 : // Headers of a streamed response, as Node gives them: names in lowercase, and the values of
15 : // a header repeated joined with ", "
16 154 : void set_headers(stream_result& result, const http_response& response) {
17 154 : result.headers.clear();
18 433 : for (const auto& [name, value] : response.get_headers()) {
19 279 : std::string key = name;
20 279 : ::thinger::util::ascii::to_lower(key);
21 279 : auto [header, inserted] = result.headers.emplace(std::move(key), value);
22 279 : if (!inserted) header->second += ", " + value;
23 279 : }
24 154 : }
25 :
26 : // Decode the Content-Encoding (gzip, x-gzip, deflate, or several of them; see
27 : // util::decode_content), up to `max_size` bytes once decoded: inflating stops as soon as the
28 : // body goes past them, so a small body expanding to gigabytes (a decompression bomb) takes no
29 : // more memory than the limit. Bodies with other codings (as Node's fetch), and empty bodies
30 : // (e.g. of responses to HEAD), are kept as they are, with their Content-Encoding.
31 6927 : client_error decompress(http_response& response, size_t max_size) {
32 6927 : if (response.get_content().empty() || !response.has_header("Content-Encoding")) return client_error::none;
33 : // The values of every Content-Encoding header make one list
34 254 : std::string encoding;
35 508 : for (const auto& value : response.get_headers_with_key("Content-Encoding")) {
36 254 : if (!encoding.empty()) encoding += ", ";
37 254 : encoding += value;
38 254 : }
39 254 : switch (::thinger::util::decode_content(encoding, response.get_content(), max_size)) {
40 173 : case ::thinger::util::content_decoding::decoded:
41 173 : response.remove_headers("Content-Encoding");
42 173 : return client_error::none;
43 9 : case ::thinger::util::content_decoding::unsupported:
44 9 : return client_error::none;
45 30 : case ::thinger::util::content_decoding::too_large:
46 30 : LOG_ERROR("the decompressed {} response exceeds the maximum allowed size: {} max", encoding, max_size);
47 30 : return client_error::body_too_large;
48 42 : case ::thinger::util::content_decoding::invalid:
49 42 : break;
50 : }
51 42 : LOG_ERROR("Failed to decompress {} response", encoding);
52 42 : return client_error::bad_content_encoding;
53 254 : }
54 :
55 6703 : std::string_view trim(std::string_view value) {
56 : static constexpr std::string_view whitespace = " \t";
57 6703 : auto first = value.find_first_not_of(whitespace);
58 6703 : if (first == std::string_view::npos) return {};
59 102 : return value.substr(first, value.find_last_not_of(whitespace) - first + 1);
60 : }
61 :
62 : // CONNECT host:port, in authority-form (RFC 9110 section 9.3.6, RFC 9112 section 3.2.3),
63 : // with the credentials of the proxy: only it sees them
64 48 : void set_connect_request(http_request& connect, const http_request& request, const std::string& authority,
65 : const std::string& proxy_authorization) {
66 48 : connect.set_method(method::CONNECT);
67 48 : connect.set_uri(authority);
68 48 : connect.add_header(header::host, authority);
69 48 : if (request.has_header(header::user_agent)) {
70 48 : connect.add_header(header::user_agent, request.get_header(header::user_agent));
71 : }
72 48 : if (!proxy_authorization.empty()) {
73 117 : connect.add_header("Proxy-Authorization", proxy_authorization);
74 : }
75 48 : }
76 :
77 : // Whether a Connection header lists `option` (e.g. "close")
78 6613 : bool has_connection_option(std::string_view value, std::string_view option) {
79 : while (true) {
80 6613 : auto comma = value.find(',');
81 6613 : if (::thinger::util::ascii::iequals(trim(value.substr(0, comma)), option)) return true;
82 6604 : if (comma == std::string_view::npos) return false;
83 0 : value.remove_prefix(comma + 1);
84 0 : }
85 : }
86 :
87 : // The timeout of a Keep-Alive header ("timeout=5, max=100"), in seconds
88 45 : std::optional<std::chrono::seconds> keep_alive_timeout_of(std::string_view value) {
89 45 : while (!value.empty()) {
90 45 : auto comma = value.find(',');
91 45 : auto parameter = trim(value.substr(0, comma));
92 45 : if (::thinger::util::ascii::istarts_with(parameter, "timeout=")) {
93 45 : auto number = trim(parameter.substr(8));
94 45 : long seconds = 0;
95 45 : auto [end, ec] = std::from_chars(number.data(), number.data() + number.size(), seconds);
96 45 : if (ec == std::errc{} && end == number.data() + number.size() && seconds >= 0) {
97 45 : return std::chrono::seconds{seconds};
98 : }
99 0 : return std::nullopt;
100 : }
101 0 : if (comma == std::string_view::npos) break;
102 0 : value.remove_prefix(comma + 1);
103 : }
104 0 : return std::nullopt;
105 : }
106 :
107 : }
108 :
109 : std::atomic<unsigned long> client_connection::connections(0);
110 :
111 15028 : client_connection::client_connection(std::shared_ptr<thinger::asio::socket> socket,
112 15028 : std::chrono::seconds timeout)
113 15028 : : socket_(std::move(socket))
114 15028 : , timeout_(timeout)
115 30056 : , idle_timer_(socket_->get_io_context()) {
116 15028 : ++connections;
117 30056 : LOG_TRACE("created http client connection with timeout: {} seconds. total: {}",
118 : timeout.count(), connections.load());
119 15028 : }
120 :
121 483 : client_connection::client_connection(std::shared_ptr<thinger::asio::unix_socket> socket,
122 : const std::string& path,
123 483 : std::chrono::seconds timeout)
124 483 : : socket_(std::move(socket))
125 483 : , socket_path_(path)
126 483 : , timeout_(timeout)
127 966 : , idle_timer_(socket_->get_io_context()) {
128 483 : ++connections;
129 966 : LOG_TRACE("created http client connection (for unix socket: {}) with timeout: {} seconds. total: {}",
130 : path, timeout.count(), connections.load());
131 483 : }
132 :
133 105 : client_connection::client_connection(boost::asio::io_context& io_context, proxy_route route,
134 105 : std::chrono::seconds timeout)
135 105 : : socket_(std::make_shared<thinger::asio::tcp_socket>("http_client", io_context))
136 105 : , route_(std::move(route))
137 105 : , timeout_(timeout)
138 315 : , idle_timer_(io_context) {
139 105 : ++connections;
140 210 : LOG_TRACE("created http client connection through proxy {} with timeout: {} seconds. total: {}",
141 : route_.proxy ? route_.proxy->url() : "", timeout.count(), connections.load());
142 105 : }
143 :
144 15616 : client_connection::~client_connection() {
145 15616 : --connections;
146 31232 : LOG_TRACE("releasing http client connection. total: {}", connections.load());
147 15616 : }
148 :
149 8516 : awaitable<boost::system::error_code> client_connection::ensure_connected(const http_request& request) {
150 : if (socket_->is_open()) {
151 : co_return boost::system::error_code{};
152 : }
153 : if (route_.proxy) {
154 : co_return co_await connect_through_proxy(request);
155 : }
156 : co_return co_await connect_socket(request.get_host(), request.get_port());
157 17032 : }
158 :
159 5093 : awaitable<boost::system::error_code> client_connection::connect_socket(const std::string& host, const std::string& port) {
160 : boost::system::error_code ec;
161 : LOG_TRACE("connecting to: {}:{}", host, port);
162 :
163 : for (unsigned retry = 0; retry < MAX_RETRIES; ++retry) {
164 : if (socket_path_.empty()) {
165 : ec = co_await socket_->connect(host, port, CONNECT_TIMEOUT);
166 : } else {
167 : auto unix_socket = std::static_pointer_cast<thinger::asio::unix_socket>(socket_);
168 : ec = co_await unix_socket->connect(socket_path_, CONNECT_TIMEOUT);
169 : }
170 :
171 : if (!ec) {
172 : LOG_TRACE("connection established");
173 : // Each request is written at once: with Nagle, a write while the previous one is
174 : // unacknowledged waits for the delayed ACK of the server (40 ms), as happens on a
175 : // reused connection. Go, Node and curl disable it too
176 : if (auto tcp = std::dynamic_pointer_cast<thinger::asio::tcp_socket>(socket_)) {
177 : tcp->enable_tcp_no_delay();
178 : }
179 : co_return ec;
180 : }
181 :
182 : LOG_ERROR("error while connecting (attempt #{}): {} ({}) ({})",
183 : retry + 1, ec.message(), ec.value(), ec.category().name());
184 :
185 : // Don't retry for certain errors, nor a failed TLS handshake (e.g. a certificate
186 : // that cannot be verified): it would fail the same way
187 : if (ec == boost::asio::error::operation_aborted ||
188 : ec == boost::asio::error::host_not_found ||
189 : ec.category() == boost::asio::error::get_ssl_category() ||
190 : ec.category() == boost::asio::ssl::error::get_stream_category()) {
191 : co_return ec;
192 : }
193 :
194 : if (retry + 1 >= MAX_RETRIES) {
195 : co_return ec;
196 : }
197 :
198 : // Close and retry
199 : socket_->close();
200 : }
201 : co_return ec;
202 10186 : }
203 :
204 105 : awaitable<boost::system::error_code> client_connection::connect_through_proxy(const http_request& request) {
205 : // A new TCP socket every time: the previous one may have become a TLS stream
206 : auto tcp = std::make_shared<thinger::asio::tcp_socket>("http_client", socket_->get_io_context());
207 : socket_ = tcp;
208 : if (auto ec = co_await connect_socket(route_.proxy->host, route_.proxy->port)) {
209 : co_return ec;
210 : }
211 : if (!route_.tunnel) {
212 : co_return boost::system::error_code{};
213 : }
214 : if (auto ec = co_await open_tunnel(request)) {
215 : co_return ec;
216 : }
217 : if (!route_.tls) {
218 : co_return boost::system::error_code{};
219 : }
220 :
221 : // TLS with the origin, end to end through the tunnel: SNI and the certificate verified
222 : // for the name of the origin, not of the proxy
223 : auto tls = std::make_shared<thinger::asio::ssl_socket>("http_client", tcp, route_.tls);
224 : socket_ = tls;
225 : auto ec = co_await tls->handshake(request.get_host());
226 : if (ec) {
227 : LOG_ERROR("TLS handshake with {} through the proxy {} failed: {}",
228 : request.get_host(), route_.proxy->url(), ec.message());
229 : }
230 : co_return ec;
231 210 : }
232 :
233 48 : awaitable<boost::system::error_code> client_connection::open_tunnel(const http_request& request) {
234 : const auto& host = request.get_host();
235 : std::string authority = (host.find(':') != std::string::npos ? "[" + host + "]" : host) + ":" + request.get_port();
236 : http_request connect;
237 : set_connect_request(connect, request, authority, route_.proxy->authorization);
238 : LOG_DEBUG("opening a tunnel to {} through the proxy {}", authority, route_.proxy->url());
239 :
240 : std::vector<boost::asio::const_buffer> buffers;
241 : connect.to_buffer(buffers);
242 : if (auto [ec, bytes] = co_await socket_->write(buffers); ec) {
243 : co_return ec;
244 : }
245 :
246 : // Its answer is parsed as strictly as any response: a 2xx to CONNECT has no body, and
247 : // the tunnel starts right after it
248 : response_factory parser;
249 : parser.set_request_method(method::CONNECT);
250 : parser.set_max_content_size(MAX_PROXY_RESPONSE_SIZE);
251 : while (true) {
252 : auto [ec, bytes] = co_await socket_->read_some(buffer_, MAX_BUFFER_SIZE);
253 :
254 : uint8_t* begin = buffer_;
255 : uint8_t* end = buffer_ + bytes;
256 : boost::tribool result = boost::indeterminate;
257 : if (begin != end) {
258 : result = parser.parse_some(begin, end);
259 : }
260 : if (boost::indeterminate(result) && ec) {
261 : if (ec != boost::asio::error::eof) co_return ec;
262 : result = parser.end_of_stream();
263 : }
264 : if (!result) {
265 : LOG_WARNING("invalid answer to CONNECT from the proxy {}: {}", route_.proxy->url(),
266 : make_error_code(parser.error()).message());
267 : co_return parser.error();
268 : }
269 : if (boost::indeterminate(result)) continue;
270 :
271 : auto response = parser.consume_response();
272 : int status = response->get_status_code();
273 : if (status >= 200 && status < 300) {
274 : // Nothing comes from the origin before the client speaks (TLS, or its request)
275 : if (begin != end) co_return client_error::proxy_tunnel_failed;
276 : co_return boost::system::error_code{};
277 : }
278 : LOG_WARNING("the proxy {} refused a tunnel to {}: {}", route_.proxy->url(), authority, status);
279 : proxy_response_ = std::move(response);
280 : co_return status == 407 ? client_error::proxy_authentication_required : client_error::proxy_tunnel_failed;
281 : }
282 96 : }
283 :
284 66 : awaitable<io_result> client_connection::write_to_proxy(const http_request& request) {
285 : // The absolute-form target names the origin (RFC 9112 section 3.2.2). The proxy consumes
286 : // its credentials (RFC 9110 section 11.7.2): unless the request has its own, they are
287 : // written here, never added to the request
288 : std::string target = request.get_url();
289 : std::string credentials;
290 : if (!route_.proxy->authorization.empty() && !request.has_header("Proxy-Authorization")) {
291 : credentials = "Proxy-Authorization: " + route_.proxy->authorization + "\r\n";
292 : }
293 : std::vector<boost::asio::const_buffer> buffers;
294 : buffers.reserve(32);
295 : request.to_buffer(buffers, target, credentials);
296 : co_return co_await socket_->write(buffers);
297 132 : }
298 :
299 8052 : awaitable<std::shared_ptr<http_response>> client_connection::read_response(const http_request& request) {
300 : while (true) {
301 : auto [ec, bytes] = co_await socket_->read_some(buffer_, MAX_BUFFER_SIZE);
302 : if (bytes > 0) response_started_ = true;
303 :
304 : uint8_t* begin = buffer_;
305 : uint8_t* end = buffer_ + bytes;
306 : boost::tribool result = boost::indeterminate;
307 : if (begin != end) {
308 : result = response_parser_.parse_some(begin, end);
309 : }
310 : if (boost::indeterminate(result) && ec) {
311 : // The end of the connection ends a body delimited by it, and nothing else
312 : if (ec != boost::asio::error::eof) {
313 : error_ = ec;
314 : co_return nullptr;
315 : }
316 : result = response_parser_.end_of_stream();
317 : }
318 : if (!result) {
319 : error_ = response_parser_.error();
320 : LOG_WARNING("invalid response from {}: {}", request.get_host(), error_.message());
321 : co_return nullptr;
322 : }
323 : if (boost::indeterminate(result)) continue;
324 : co_return complete_response(request, begin, end);
325 : }
326 16104 : }
327 :
328 7010 : std::shared_ptr<http_response> client_connection::complete_response(const http_request& request,
329 : const uint8_t* begin, const uint8_t* end) {
330 : // Data past the response, other than that of a switched protocol (another response
331 : // than the one requested, or garbage), leaves the connection out of sync. A client
332 : // asking to close it sends nothing more on it (RFC 9112 section 9.6)
333 13623 : bool reusable = response_parser_.keep_alive() && begin == end &&
334 6613 : !has_connection_option(request.get_header(header::connection), "close");
335 7010 : auto response = response_parser_.consume_response();
336 7010 : if (reusable) {
337 6604 : reusable_ = true;
338 6604 : if (response->has_header("Keep-Alive")) {
339 45 : keep_alive_timeout_ = keep_alive_timeout_of(response->get_header("Keep-Alive"));
340 : }
341 : }
342 7010 : if (response->get_status_code() == 101) {
343 271 : if (!request.has_header(header::upgrade)) {
344 6 : error_ = client_error::unexpected_upgrade;
345 6 : return nullptr;
346 : }
347 : // What followed the response belongs to the new protocol, e.g. WebSocket frames
348 265 : buffered_.assign(reinterpret_cast<const char*>(begin), static_cast<size_t>(end - begin));
349 6739 : } else if (!reusable) {
350 135 : socket_->close();
351 : }
352 :
353 : // A streamed body is handed as received. A body that cannot be decoded fails the
354 : // request, as one too large once decoded, and the connection is closed as after any
355 : // error
356 7004 : if (decompress_ && !response_parser_.is_streaming()) {
357 6927 : if (auto error = decompress(*response, response_parser_.get_max_content_size()); error != client_error::none) {
358 72 : error_ = error;
359 72 : return nullptr;
360 : }
361 : }
362 6932 : return response;
363 7010 : }
364 :
365 373 : awaitable<std::shared_ptr<http_response>> client_connection::read_after_failed_write(
366 : const http_request& request, boost::system::error_code write_ec, size_t bytes) {
367 : error_ = write_ec;
368 : request_unsent_ = bytes == 0;
369 : // The server may have answered before taking the whole request, and closed the
370 : // connection (e.g. 413 for a body over its limit): its answer may be there to read,
371 : // as curl does. The connection is never reused then. A Unix socket whose peer closed fails
372 : // the write with not_connected on macOS
373 : if (bytes == 0 || (write_ec != boost::asio::error::broken_pipe &&
374 : write_ec != boost::asio::error::connection_reset &&
375 : write_ec != boost::asio::error::not_connected)) {
376 : co_return nullptr;
377 : }
378 : auto response = co_await read_response(request);
379 : if (!response || response->get_status_code() == 101) {
380 : error_ = write_ec;
381 : co_return nullptr;
382 : }
383 : // A 408 on a reused connection is most likely the server closing it while idle,
384 : // before this request arrived: not its answer, and it may be sent again (see
385 : // http_client_base::can_retry()), as Go does
386 : if (reused_ && response->get_status_code() == 408) {
387 : error_ = write_ec;
388 : response_started_ = false;
389 : request_unsent_ = true;
390 : co_return nullptr;
391 : }
392 : error_ = {};
393 : reusable_ = false;
394 : socket_->close();
395 : co_return response;
396 746 : }
397 :
398 8516 : awaitable<std::shared_ptr<http_response>> client_connection::exchange(std::shared_ptr<http_request> request) {
399 : std::shared_ptr<http_response> response;
400 : bool timed_out = false;
401 : error_ = {};
402 : buffered_.clear();
403 : proxy_response_.reset();
404 : reused_ = socket_->is_open();
405 : request_unsent_ = false;
406 : response_started_ = false;
407 : reusable_ = false;
408 : keep_alive_timeout_.reset();
409 :
410 : // Timeout timer
411 : boost::asio::steady_timer timer(socket_->get_io_context());
412 : timer.expires_after(timeout_);
413 :
414 8516 : auto send_and_read = [&]() -> awaitable<void> {
415 : if (auto ec = co_await ensure_connected(*request)) {
416 : error_ = ec;
417 : co_return;
418 : }
419 : if (!socket_->is_open()) {
420 : error_ = boost::asio::error::not_connected;
421 : co_return;
422 : }
423 :
424 : request->log("CLIENT->", body_logging_);
425 : io_result written;
426 : if (route_.proxy && !route_.tunnel) {
427 : written = co_await write_to_proxy(*request);
428 : } else {
429 : written = co_await request->to_socket(socket_);
430 : }
431 : if (auto [write_ec, bytes] = written; write_ec) {
432 : response = co_await read_after_failed_write(*request, write_ec, bytes);
433 : co_return;
434 : }
435 : response = co_await read_response(*request);
436 17032 : };
437 :
438 : // A cancellation that finds the request between two operations (one completed, its
439 : // continuation queued or running) does not end an operation with operation_aborted: the
440 : // next co_await throws it. The error is kept here, as || takes the end of do_timeout,
441 : // cancelled at the same time, over an exception
442 8516 : auto do_request = [&]() -> awaitable<void> {
443 : try {
444 : co_await send_and_read();
445 : } catch (const boost::system::system_error& error) {
446 : error_ = error.code();
447 : }
448 17032 : };
449 :
450 8516 : auto do_timeout = [&]() -> awaitable<void> {
451 : auto [ec] = co_await timer.async_wait(use_nothrow_awaitable);
452 : if (!ec) {
453 : LOG_ERROR("Request timeout after {} seconds", timeout_.count());
454 : timed_out = true;
455 : socket_->close();
456 : }
457 17032 : };
458 :
459 : co_await (do_request() || do_timeout());
460 :
461 : if (timed_out) {
462 : response.reset();
463 : error_ = boost::asio::error::timed_out;
464 : }
465 : // Never reused after an error: where the next response would start is unknown
466 : if (!response) {
467 : reusable_ = false;
468 : close();
469 : }
470 :
471 : co_return response;
472 17032 : }
473 :
474 8406 : awaitable<std::shared_ptr<http_response>> client_connection::send_request(
475 : std::shared_ptr<http_request> request) {
476 : response_parser_.reset();
477 : response_parser_.set_request_method(request->get_method());
478 : response_parser_.setOnChunked(request->get_chunked_callback());
479 : co_return co_await exchange(std::move(request));
480 16812 : }
481 :
482 110 : awaitable<stream_result> client_connection::send_request_streaming(
483 : std::shared_ptr<http_request> request,
484 : stream_callback callback) {
485 :
486 : stream_result result;
487 :
488 : response_parser_.reset();
489 : response_parser_.set_request_method(request->get_method());
490 : response_parser_.setOnStreaming(
491 20822 : [&callback, &result, this](
492 : const std::string_view& data, size_t downloaded, size_t total) -> bool {
493 20822 : if (result.status_code == 0) {
494 86 : result.status_code = response_parser_.get_status_code();
495 : // The head is parsed: the callback may stop the download before its end
496 86 : if (auto head = response_parser_.get_response()) set_headers(result, *head);
497 : }
498 20822 : result.bytes_transferred = downloaded;
499 :
500 20822 : stream_info info{data, downloaded, total, result.status_code};
501 41644 : return callback(info);
502 : });
503 :
504 : auto response = co_await exchange(std::move(request));
505 : if (!response) {
506 : result.error = error_.message();
507 : // The status of a proxy refusing the tunnel, with its error
508 : if (proxy_response_) {
509 : result.status_code = proxy_response_->get_status_code();
510 : set_headers(result, *proxy_response_);
511 : }
512 : co_return result;
513 : }
514 :
515 : result.status_code = response->get_status_code();
516 : set_headers(result, *response);
517 : if (result.bytes_transferred == 0) {
518 : result.bytes_transferred = response->get_content_size();
519 : }
520 : co_return result;
521 220 : }
522 :
523 10503 : void client_connection::close() {
524 10503 : if (socket_->is_open()) {
525 4539 : socket_->close();
526 : }
527 10503 : response_parser_.reset();
528 10503 : }
529 :
530 47543 : void client_connection::expire_after(std::chrono::steady_clock::duration timeout, std::function<void()> on_expired) {
531 47543 : idle_timer_.expires_after(timeout);
532 47543 : idle_timer_.async_wait([on_expired = std::move(on_expired)](const boost::system::error_code& ec) {
533 4181 : if (!ec) on_expired();
534 4181 : });
535 47543 : }
536 :
537 262 : std::string client_connection::take_buffered() {
538 262 : return std::exchange(buffered_, {});
539 : }
540 :
541 262 : std::shared_ptr<thinger::asio::socket> client_connection::release_socket() {
542 262 : socket_->cancel();
543 262 : return socket_;
544 : }
545 :
546 : }
|
