Line data Source code
1 : // The reverse proxy, and the course of an exchange: the connection to the target, and the
2 : // timeouts. The request, its body, the response and the tunnels are in reverse_proxy/.
3 : #include "reverse_proxy/proxy_exchange.hpp"
4 : #include "server_connection.hpp"
5 : #include "../../asio/io_context_of.hpp"
6 : #include "../../asio/sockets/ssl_socket.hpp"
7 : #include "../../asio/sockets/tcp_socket.hpp"
8 : #include "../../asio/sockets/unix_socket.hpp"
9 : #include "../../util/logger.hpp"
10 : #include <boost/asio/ssl/error.hpp>
11 : #include <openssl/ssl.h>
12 : #include <openssl/x509v3.h>
13 : #include <algorithm>
14 :
15 : namespace thinger::http {
16 :
17 : namespace proxy_detail {
18 :
19 1326 : boost::asio::io_context* proxy_exchange::io_context() {
20 1326 : if (auto* io = req_.source_ ? req_.source_->io_context() : nullptr) return io;
21 : // A request dispatched in memory: the context running the handler, if it is an
22 : // io_context (null otherwise: the sockets need one)
23 6 : return asio::io_context_of(executor_);
24 : }
25 :
26 141 : std::string proxy_exchange::describe() const {
27 141 : std::string where;
28 141 : switch (target_.type) {
29 3 : case proxy_target::transport::unix_socket:
30 3 : where = "unix:" + target_.socket_path;
31 3 : break;
32 9 : case proxy_target::transport::tls:
33 9 : where = "https://" + target_.authority();
34 9 : break;
35 129 : default:
36 129 : where = "http://" + target_.authority();
37 129 : break;
38 : }
39 : // Without the query, as the other logs of a request being handled: it may have secrets
40 282 : return get_method(incoming_->get_method()) + " " + incoming_->get_path() + " -> " + where;
41 141 : }
42 :
43 153 : void proxy_exchange::fail(http_response::status status, const std::string& message, const std::string& detail) {
44 : // Once stopped (timeout, client gone, invalid request body), what the exchange waits
45 : // for fails as a consequence: that is answered by whoever stopped it
46 153 : if (stopped_ != stop_reason::none) return;
47 111 : LOG_WARNING("reverse proxy {}: {}", describe(), detail);
48 111 : if (result_.error.empty()) result_.error = detail;
49 111 : result_.complete = false;
50 111 : if (response_started_) {
51 9 : res_.abort_stream();
52 9 : return;
53 : }
54 102 : if (res_.has_responded()) return;
55 99 : res_.error(status, message);
56 99 : result_.status = static_cast<int>(status);
57 99 : result_.from_target = false;
58 : }
59 :
60 566 : void proxy_exchange::close_target() {
61 566 : if (target_socket_) target_socket_->close();
62 566 : }
63 :
64 : // --- Connection ---
65 :
66 663 : awaitable<boost::system::error_code> proxy_exchange::connect_target() {
67 : auto& io = *io_context();
68 : // The connect timeout is enforced by connect(): this one only bounds the socket
69 : auto limit = std::chrono::ceil<std::chrono::seconds>(options_.connect_timeout) + std::chrono::seconds(1);
70 :
71 : if (target_.type == proxy_target::transport::unix_socket) {
72 : auto socket = std::make_shared<asio::unix_socket>("reverse_proxy", io);
73 : target_socket_ = socket;
74 : co_return co_await socket->connect(target_.socket_path, limit);
75 : }
76 :
77 : auto tcp = std::make_shared<asio::tcp_socket>("reverse_proxy", io);
78 : target_socket_ = tcp;
79 : auto ec = co_await tcp->connect(target_.host, std::to_string(target_.port), limit);
80 : if (ec) co_return ec;
81 : tcp->enable_tcp_no_delay();
82 : if (target_.type != proxy_target::transport::tls) co_return ec;
83 : co_return co_await start_tls(std::move(tcp));
84 1326 : }
85 :
86 114 : awaitable<boost::system::error_code> proxy_exchange::start_tls(std::shared_ptr<asio::tcp_socket> tcp) {
87 : std::shared_ptr<boost::asio::ssl::context> context;
88 : try {
89 : context = proxy_.tls_context(target_.tls);
90 : } catch (const std::exception& e) {
91 : LOG_ERROR("reverse proxy: invalid TLS settings for {}: {}", target_.authority(), e.what());
92 : tls_config_error_ = true;
93 : co_return boost::asio::error::invalid_argument;
94 : }
95 :
96 : auto tls = std::make_shared<asio::ssl_socket>("reverse_proxy", tcp, context);
97 : target_socket_ = tls;
98 : const auto& name = target_.tls.server_name.empty() ? target_.host : target_.tls.server_name;
99 : if (target_.tls.verify) {
100 : // The certificate must be issued for the name (or address) of the target
101 : auto* ssl = tls->native_handle();
102 : int set = is_ip_address(name) ? X509_VERIFY_PARAM_set1_ip_asc(SSL_get0_param(ssl), name.c_str())
103 : : SSL_set1_host(ssl, name.c_str());
104 : if (set != 1) {
105 : LOG_ERROR("reverse proxy: cannot verify the name {} of {}", name, target_.authority());
106 : tls_config_error_ = true;
107 : co_return boost::asio::error::invalid_argument;
108 : }
109 : }
110 : co_return co_await tls->handshake(name);
111 228 : }
112 :
113 663 : awaitable<bool> proxy_exchange::connect() {
114 : auto limit = options_.connect_timeout;
115 : if (options_.total_timeout.count() > 0) limit = std::min(limit, options_.total_timeout);
116 :
117 : // On timeout the socket is closed, so the connection or handshake in progress fails
118 : boost::asio::steady_timer timer(executor_, limit);
119 663 : auto timeout = [&]() -> awaitable<void> {
120 : auto [ec] = co_await timer.async_wait(use_nothrow_awaitable);
121 : if (!ec) close_target();
122 1326 : };
123 : auto result = co_await (connect_target() || timeout());
124 :
125 : boost::system::error_code ec = boost::asio::error::timed_out;
126 : if (result.index() == 0) ec = std::get<0>(result);
127 : if (!ec) {
128 : touch();
129 : co_return true;
130 : }
131 :
132 : close_target();
133 : target_socket_.reset();
134 : if (ec == boost::asio::error::timed_out) {
135 : fail(http_response::status::gateway_timeout, "The target did not answer in time",
136 : "timeout connecting to the target");
137 : } else if (tls_config_error_) {
138 : fail(http_response::status::bad_gateway, "Cannot connect to the target", "invalid TLS settings");
139 : } else {
140 : fail(http_response::status::bad_gateway, "Cannot connect to the target",
141 : "cannot connect to the target: " + ec.message());
142 : }
143 : co_return false;
144 1326 : }
145 :
146 : // --- Exchange ---
147 :
148 633 : awaitable<void> proxy_exchange::exchange() {
149 : bool target_open = co_await send_head();
150 : if (target_open && body_mode_ != body_mode::none && !body_buffered_) {
151 : // The response is read while the body is sent: a target may answer before reading
152 : // it all (e.g. 401, 413), and stop reading it. Which one ends first does not tell:
153 : // the body waits for the head, which is there if it was read
154 : co_await (read_head_while_streaming() || stream_body());
155 : if (stopped_ != stop_reason::none) co_return;
156 : if (!head_) co_return;
157 : // Answered before the end of the body: what is left of it is not read
158 : if (!request_sent_) close_client_after_response();
159 : } else {
160 : if (target_open) request_done();
161 : if (!co_await read_head()) co_return;
162 : }
163 : if (status_code_ == 101) {
164 : upgrade();
165 : co_return;
166 : }
167 : co_await relay_response();
168 1266 : }
169 :
170 156 : awaitable<bool> proxy_exchange::read_head_while_streaming() {
171 : bool read = co_await read_head();
172 : head_read_->set();
173 : co_return read;
174 312 : }
175 :
176 633 : awaitable<void> proxy_exchange::watch_client() {
177 : // The scope of the request ends the exchange once cancelled: its client is gone (it reset
178 : // the connection, or a read or write on it failed; not a client that only closed its
179 : // sending side), its route timed out, or the server reached its shutdown deadline. Nobody
180 : // would take the response then (e.g. an event stream that stays silent). The connection of
181 : // the client is not read here: the server watches it
182 : auto reason = co_await req_.client_gone();
183 : // The exchange ended meanwhile: the wait was cancelled
184 : if (reason == cancel_reason::none) co_return;
185 : // The whole response is queued: the exchange is done with the client, and ends once it is
186 : // written
187 : if (response_ended_) {
188 : co_await response_written_->wait();
189 : co_return;
190 : }
191 : if (stopped_ == stop_reason::none) {
192 : stopped_ = reason == cancel_reason::client_gone ? stop_reason::client_left : stop_reason::timeout;
193 : }
194 : close_target();
195 : // A response started is incomplete: the connection of the client is closed after it
196 : if (reason != cancel_reason::client_gone && response_started_) {
197 : if (auto connection = res_.get_connection()) connection->release();
198 : }
199 1266 : }
200 :
201 633 : bool proxy_exchange::has_deadline() const {
202 633 : return options_.idle_timeout.count() > 0 || options_.total_timeout.count() > 0;
203 : }
204 :
205 : // The earliest of the idle and total deadlines, or none when neither timeout is set
206 24961 : std::optional<clock::time_point> proxy_exchange::next_deadline() const {
207 24961 : std::optional<clock::time_point> deadline;
208 24961 : if (options_.idle_timeout.count() > 0) deadline = last_activity_ + options_.idle_timeout;
209 24961 : if (options_.total_timeout.count() > 0) {
210 6 : auto total = start_ + options_.total_timeout;
211 6 : deadline = deadline ? std::min(*deadline, total) : total;
212 : }
213 24961 : return deadline;
214 : }
215 :
216 24991 : void proxy_exchange::touch() {
217 24991 : last_activity_ = clock::now();
218 : // Later: the watchdog checks it when it expires
219 24991 : if (deadline_) deadline_->expires_at(*next_deadline());
220 24991 : }
221 :
222 621 : void proxy_exchange::start_deadline() {
223 621 : deadline_passed_.emplace(executor_);
224 630 : deadline_.emplace(executor_, [this] { deadline_passed_->set(); });
225 621 : deadline_->expires_at(*next_deadline());
226 621 : }
227 :
228 621 : awaitable<void> proxy_exchange::stop_at_deadline() {
229 : // Cancelled: the exchange ended
230 : if (!co_await deadline_passed_->wait()) co_return;
231 : stop_for_timeout();
232 1242 : }
233 :
234 9 : void proxy_exchange::stop_for_timeout() {
235 : // Make whatever the exchange waits for fail: the target, and a client that does
236 : // not take the response (its connection is closed, as the response is incomplete)
237 9 : if (stopped_ == stop_reason::none) stopped_ = stop_reason::timeout;
238 9 : close_target();
239 9 : if (response_started_) {
240 3 : if (auto connection = res_.get_connection()) connection->release();
241 : }
242 9 : }
243 :
244 488 : void proxy_exchange::finish_stopped() {
245 488 : switch (std::exchange(stopped_, stop_reason::none)) {
246 12 : case stop_reason::timeout:
247 12 : if (response_started_) {
248 3 : result_.complete = false;
249 3 : result_.error = "timeout relaying the response";
250 3 : LOG_WARNING("reverse proxy {}: {}", describe(), result_.error);
251 3 : res_.abort_stream();
252 : } else {
253 : // what the exchange failed with is the consequence of the timeout
254 9 : result_.error.clear();
255 36 : fail(http_response::status::gateway_timeout, "The target did not answer in time",
256 : "timeout waiting for the target");
257 : }
258 12 : break;
259 12 : case stop_reason::client_left:
260 12 : result_.complete = false;
261 12 : result_.error = "the client closed the connection";
262 12 : LOG_DEBUG("reverse proxy {}: {}", describe(), result_.error);
263 12 : if (response_started_) res_.abort_stream();
264 12 : break;
265 464 : default:
266 464 : break;
267 : }
268 488 : }
269 :
270 675 : awaitable<proxy_result> proxy_exchange::run() {
271 : executor_ = co_await boost::asio::this_coro::executor;
272 : start_ = clock::now();
273 : last_activity_ = start_;
274 : if (!incoming_) {
275 : result_.error = "no request";
276 : co_return result_;
277 : }
278 : if (!prepare()) co_return result_;
279 : if (!io_context()) {
280 : fail(http_response::status::not_implemented, "", "the handler does not run on an io_context");
281 : co_return result_;
282 : }
283 : if (!co_await connect()) co_return result_;
284 :
285 : head_read_.emplace(executor_);
286 : response_written_.emplace(executor_);
287 : if (has_deadline()) {
288 : start_deadline();
289 : co_await (exchange() || stop_at_deadline() || watch_client());
290 : deadline_.reset();
291 : } else {
292 : // No timeout: nothing to watch but the client
293 : co_await (exchange() || watch_client());
294 : }
295 : finish_stopped();
296 : close_target();
297 : co_return result_;
298 1350 : }
299 :
300 : }
301 :
302 : // --- reverse_proxy ---
303 :
304 114 : std::shared_ptr<boost::asio::ssl::context> reverse_proxy::tls_context(const proxy_tls_options& tls) {
305 114 : auto key = std::make_tuple(tls.verify, tls.ca_file, tls.ca_pem);
306 114 : std::lock_guard lock(tls_mutex_);
307 114 : if (auto it = tls_contexts_.find(key); it != tls_contexts_.end()) return it->second;
308 :
309 96 : auto context = std::make_shared<boost::asio::ssl::context>(boost::asio::ssl::context::tls_client);
310 96 : context->set_options(boost::asio::ssl::context::default_workarounds | boost::asio::ssl::context::no_sslv2 |
311 : boost::asio::ssl::context::no_sslv3 | boost::asio::ssl::context::no_tlsv1 |
312 : boost::asio::ssl::context::no_tlsv1_1);
313 96 : if (tls.verify) {
314 90 : context->set_verify_mode(boost::asio::ssl::verify_peer);
315 90 : if (!tls.ca_file.empty()) {
316 0 : context->load_verify_file(tls.ca_file);
317 90 : } else if (!tls.ca_pem.empty()) {
318 87 : context->add_certificate_authority(boost::asio::buffer(tls.ca_pem));
319 : } else {
320 3 : context->set_default_verify_paths();
321 : }
322 : } else {
323 6 : context->set_verify_mode(boost::asio::ssl::verify_none);
324 : }
325 96 : tls_contexts_.emplace(key, context);
326 96 : return context;
327 114 : }
328 :
329 588 : awaitable<proxy_result> reverse_proxy::forward(request& req, response& res, const proxy_target& target) {
330 : co_return co_await forward(req, res, target, defaults_);
331 1176 : }
332 :
333 675 : awaitable<proxy_result> reverse_proxy::forward(request& req, response& res, const proxy_target& target,
334 : const proxy_options& options) {
335 : proxy_detail::proxy_exchange exchange(*this, req, res, target, options);
336 : co_return co_await exchange.run();
337 1350 : }
338 :
339 : } // namespace thinger::http
|
