1018307341

Committed 27 Sep 2023 12:04PM UTC coverage: 77.682% (-0.4%) from 78.076%

Build # 1018307341

Build Type

push

gitlab-ci

Committed by

vpodzime

Commit Message

fix: Wait repeatedly for I/O events on DBus file descriptors

ASIO's `stream_descriptor::async_read/wait/error()` all register
one-off handlers. We actually need to watch the given file
descriptors for all events until we cancel the watch.

We can do that by using functors re-registering themselves
whenever called.

Also, add a tip and commented code to DBus tests helping debug
DBus issues.

Ticket: MEN-6655
Changelog: none
Signed-off-by: Vratislav Podzimek <v.podzimek@mykolab.com>

Run Details

17 of 17 new or added lines in 1 file covered. (100.0%)

6474 of 8334 relevant lines covered (77.68%)

11035.56 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.74

/mender-auth/http_forwarder/http_forwarder.cpp

// Copyright 2023 Northern.tech AS
//
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//
//        http://www.apache.org/licenses/LICENSE-2.0
//
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.

#include <mender-auth/http_forwarder.hpp>

namespace mender {
namespace auth {
namespace http_forwarder {

ForwardObject::ForwardObject(const http::ClientConfig &config, events::EventLoop &event_loop) :
        client_(config, event_loop),
        logger_("http_forwarder") {
}

Server::Server(
        const http::ServerConfig &server_config,
        const http::ClientConfig &client_config,
        events::EventLoop &loop) :
        logger_("http_forwarder"),
        event_loop_ {loop},
        server_ {server_config, loop},
        cancelled_ {make_shared<bool>(true)},
        client_config_ {client_config} {
}

Server::~Server() {
        *cancelled_ = true;
        cancelled_ = make_shared<bool>(true);
        Cancel();
}

void Server::Cancel() {
        *cancelled_ = true;
        cancelled_ = make_shared<bool>(true);
        connections_.clear();
        server_.Cancel();
}

error::Error Server::AsyncForward(const string &listen_url, const string &target_url) {
        if (!*cancelled_) {
                return error::Error(
                        make_error_condition(errc::operation_in_progress),
                        "HTTP forwarding already in progress");
        }

        *cancelled_ = false;

        http::BrokenDownUrl target_address;

        // We don't actually need target_address here, but break it down here anyway to avoid that
        // the error shows up much later when the first connection is made.
        auto err = http::BreakDownUrl(target_url, target_address);
        if (err != error::NoError) {
                return err.WithContext("HTTP forwarder: Invalid target address");
        }
        target_url_ = target_url;

        auto &cancelled = cancelled_;
        err = server_.AsyncServeUrl(
                listen_url,
                [this, cancelled](http::ExpectedIncomingRequestPtr exp_req) {
                        if (!*cancelled) {
                                RequestHeaderHandler(exp_req);
                        }
                },
                [this, cancelled](http::IncomingRequestPtr req, error::Error err) {
                        if (!*cancelled) {
                                RequestBodyHandler(req, err);
                        }
                });
        if (err != error::NoError) {
                return err.WithContext("Unable to start HTTP forwarding server");
        }

        return error::NoError;
}

uint16_t Server::GetPort() const {
        return server_.GetPort();
}

string Server::GetUrl() const {
        return server_.GetUrl();
}

void Server::RequestHeaderHandler(http::ExpectedIncomingRequestPtr exp_req) {
        if (!exp_req) {
                logger_.Error("Error in incoming request: " + exp_req.error().String());
                return;
        }
        auto &req_in = exp_req.value();

        ForwardObjectPtr connection {new ForwardObject(client_config_, event_loop_)};
        connections_[req_in] = connection;
        connection->logger_ = logger_.WithFields(log::LogField {"request", req_in->GetPath()});
        connection->req_in_ = req_in;

        auto final_url = http::JoinUrl(target_url_, req_in->GetPath());
        auto req_out = make_shared<http::OutgoingRequest>();
        req_out->SetMethod(req_in->GetMethod());
        req_out->SetAddress(final_url);
        for (auto header : req_in->GetHeaders()) {
                req_out->SetHeader(header.first, header.second);
        }
        connection->req_out_ = req_out;

        auto exp_body_reader = req_in->MakeBodyAsyncReader();
        if (exp_body_reader) {
                auto body_reader = exp_body_reader.value();
                auto generated = make_shared<bool>(false);
                req_out->SetAsyncBodyGenerator([body_reader, generated]() -> io::ExpectedAsyncReaderPtr {
                        // We can only do this once, because the incoming request body is not
                        // seekable.
                        if (*generated) {
                                return expected::unexpected(error::Error(
                                        make_error_condition(errc::invalid_seek),
                                        "Cannot rewind HTTP stream to regenerate body"));
                        } else {
                                *generated = true;
                                return body_reader;
                        }
                });
        } else if (exp_body_reader.error().code != http::MakeError(http::BodyMissingError, "").code) {
                connection->logger_.Error(
                        "Could not get body reader for request: " + exp_body_reader.error().String());
                connections_.erase(req_in);
                return;
        } // else: if body is missing we don't need to do anything.

        auto &cancelled = cancelled_;
        auto err = connection->client_.AsyncCall(
                req_out,
                [this, cancelled, req_in](http::ExpectedIncomingResponsePtr exp_resp) {
                        if (!*cancelled) {
                                ResponseHeaderHandler(req_in, exp_resp);
                        }
                },
                [this, cancelled, req_in](http::ExpectedIncomingResponsePtr exp_resp) {
                        if (!*cancelled) {
                                ResponseBodyHandler(req_in, exp_resp);
                        }
                });
}

void Server::RequestBodyHandler(http::IncomingRequestPtr req_in, error::Error err) {
        auto maybe_connection = connections_.find(req_in);
        if (maybe_connection == connections_.end()) {
                // Can happen if the request was cancelled.
                return;
        }
        auto &connection = maybe_connection->second;

        if (err != error::NoError) {
                connection->logger_.Error("Error while reading incoming request body: " + err.String());
                connections_.erase(req_in);
                return;
        }

        auto exp_resp_out = connection->req_in_->MakeResponse();
        if (!exp_resp_out) {
                connection->logger_.Error(
                        "Could not make outgoing response: " + exp_resp_out.error().String());
                connections_.erase(req_in);
                return;
        }
        connection->resp_out_ = exp_resp_out.value();
}

void Server::ResponseHeaderHandler(
        http::IncomingRequestPtr req_in, http::ExpectedIncomingResponsePtr exp_resp_in) {
        auto connection = connections_[req_in];

        if (!exp_resp_in) {
                connection->logger_.Error("Error in incoming response: " + exp_resp_in.error().String());
                connections_.erase(req_in);
                return;
        }
        connection->resp_in_ = exp_resp_in.value();
        auto &resp_in = connection->resp_in_;

        auto &resp_out = connection->resp_out_;

        resp_out->SetStatusCodeAndMessage(resp_in->GetStatusCode(), resp_in->GetStatusMessage());
        for (auto header : resp_in->GetHeaders()) {
                resp_out->SetHeader(header.first, header.second);
        }

        auto exp_body_reader = resp_in->MakeBodyAsyncReader();
        if (exp_body_reader) {
                resp_out->SetAsyncBodyReader(exp_body_reader.value());
        } else if (exp_body_reader.error().code != http::MakeError(http::BodyMissingError, "").code) {
                connection->logger_.Error(
                        "Could not get body reader for response: " + exp_body_reader.error().String());
                connections_.erase(req_in);
                return;
        } // else: if body is missing we don't need to do anything.

        auto &cancelled = cancelled_;
        auto err = resp_out->AsyncReply([cancelled, this, req_in](error::Error err) {
                if (*cancelled) {
                        return;
                }

                if (err != error::NoError) {
                        connections_[req_in]->logger_.Error(
                                "Error while forwarding response to client: " + err.String());
                        connections_.erase(req_in);
                        return;
                }

                auto &connection = connections_[req_in];
                connection->incoming_request_finished = true;
                if (connection->outgoing_request_finished) {
                        // We are done, remove connection.
                        connections_.erase(req_in);
                }
        });
        if (err != error::NoError) {
                connection->logger_.Error("Could not forward response to client: " + err.String());
                connections_.erase(req_in);
                return;
        }
}

void Server::ResponseBodyHandler(
        http::IncomingRequestPtr req_in, http::ExpectedIncomingResponsePtr exp_resp_in) {
        auto &connection = connections_[req_in];

        if (!exp_resp_in) {
                connection->logger_.Error(
                        "Error while reading incoming response body: " + exp_resp_in.error().String());
                connections_.erase(req_in);
                return;
        }

        connection->outgoing_request_finished = true;
        if (connection->incoming_request_finished) {
                // We are done, remove connection.
                connections_.erase(req_in);
        }
}

} // namespace http_forwarder
} // namespace auth
} // namespace mender

1	// Copyright 2023 Northern.tech AS
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	#include <mender-auth/http_forwarder.hpp>
16
17	namespace mender {
18	namespace auth {
19	namespace http_forwarder {
20
21	ForwardObject::ForwardObject(const http::ClientConfig &config, events::EventLoop &event_loop) :	7✔
22	client_(config, event_loop),
23	logger_("http_forwarder") {	7✔
24	}	7✔
25
26	Server::Server(	7✔
27	const http::ServerConfig &server_config,
28	const http::ClientConfig &client_config,
29	events::EventLoop &loop) :	×
30	logger_("http_forwarder"),
31	event_loop_ {loop},
32	server_ {server_config, loop},
33	cancelled_ {make_shared<bool>(true)},	×
34	client_config_ {client_config} {	7✔
35	}	7✔
36
37	Server::~Server() {	7✔
38	*cancelled_ = true;	7✔
39	cancelled_ = make_shared<bool>(true);	7✔
40	Cancel();	7✔
41	}	7✔
42
43	void Server::Cancel() {	8✔
44	*cancelled_ = true;	8✔
45	cancelled_ = make_shared<bool>(true);	8✔
46	connections_.clear();	8✔
47	server_.Cancel();	8✔
48	}	8✔
49
50	error::Error Server::AsyncForward(const string &listen_url, const string &target_url) {	9✔
51	if (!*cancelled_) {	9✔
52	return error::Error(
53	make_error_condition(errc::operation_in_progress),	1✔
54	"HTTP forwarding already in progress");	3✔
55	}
56
57	*cancelled_ = false;	8✔
58
59	http::BrokenDownUrl target_address;	16✔
60
61	// We don't actually need target_address here, but break it down here anyway to avoid that
62	// the error shows up much later when the first connection is made.
63	auto err = http::BreakDownUrl(target_url, target_address);	16✔
64	if (err != error::NoError) {	8✔
65	return err.WithContext("HTTP forwarder: Invalid target address");	×
66	}
67	target_url_ = target_url;	8✔
68
69	auto &cancelled = cancelled_;	8✔
70	err = server_.AsyncServeUrl(	8✔
71	listen_url,
72	[this, cancelled](http::ExpectedIncomingRequestPtr exp_req) {	14✔
73	if (!*cancelled) {	7✔
74	RequestHeaderHandler(exp_req);	7✔
75	}
76	},	7✔
77	[this, cancelled](http::IncomingRequestPtr req, error::Error err) {	14✔
78	if (!*cancelled) {	7✔
79	RequestBodyHandler(req, err);	7✔
80	}
81	});	23✔
82	if (err != error::NoError) {	8✔
83	return err.WithContext("Unable to start HTTP forwarding server");	×
84	}
85
86	return error::NoError;	8✔
87	}
88
89	uint16_t Server::GetPort() const {	×
90	return server_.GetPort();	×
91	}
92
93	string Server::GetUrl() const {	7✔
94	return server_.GetUrl();	7✔
95	}
96
97	void Server::RequestHeaderHandler(http::ExpectedIncomingRequestPtr exp_req) {	7✔
98	if (!exp_req) {	7✔
99	logger_.Error("Error in incoming request: " + exp_req.error().String());	×
100	return;	×
101	}
102	auto &req_in = exp_req.value();	7✔
103
104	ForwardObjectPtr connection {new ForwardObject(client_config_, event_loop_)};	7✔
105	connections_[req_in] = connection;	7✔
106	connection->logger_ = logger_.WithFields(log::LogField {"request", req_in->GetPath()});	7✔
107	connection->req_in_ = req_in;	7✔
108
109	auto final_url = http::JoinUrl(target_url_, req_in->GetPath());	7✔
110	auto req_out = make_shared<http::OutgoingRequest>();	7✔
111	req_out->SetMethod(req_in->GetMethod());	7✔
112	req_out->SetAddress(final_url);	7✔
113	for (auto header : req_in->GetHeaders()) {	19✔
114	req_out->SetHeader(header.first, header.second);	12✔
115	}
116	connection->req_out_ = req_out;	7✔
117
118	auto exp_body_reader = req_in->MakeBodyAsyncReader();	7✔
119	if (exp_body_reader) {	7✔
120	auto body_reader = exp_body_reader.value();	10✔
121	auto generated = make_shared<bool>(false);	5✔
122	req_out->SetAsyncBodyGenerator([body_reader, generated]() -> io::ExpectedAsyncReaderPtr {	10✔
123	// We can only do this once, because the incoming request body is not
124	// seekable.
125	if (*generated) {	5✔
126	return expected::unexpected(error::Error(	×
127	make_error_condition(errc::invalid_seek),	×
128	"Cannot rewind HTTP stream to regenerate body"));	×
129	} else {
130	*generated = true;	5✔
131	return body_reader;	5✔
132	}
133	});	10✔
134	} else if (exp_body_reader.error().code != http::MakeError(http::BodyMissingError, "").code) {	2✔
135	connection->logger_.Error(	×
136	"Could not get body reader for request: " + exp_body_reader.error().String());	×
137	connections_.erase(req_in);	×
138	return;	×
139	} // else: if body is missing we don't need to do anything.
140
141	auto &cancelled = cancelled_;	7✔
142	auto err = connection->client_.AsyncCall(	7✔
143	req_out,
144	[this, cancelled, req_in](http::ExpectedIncomingResponsePtr exp_resp) {	14✔
145	if (!*cancelled) {	7✔
146	ResponseHeaderHandler(req_in, exp_resp);	7✔
147	}
148	},	7✔
149	[this, cancelled, req_in](http::ExpectedIncomingResponsePtr exp_resp) {	4✔
150	if (!*cancelled) {	2✔
151	ResponseBodyHandler(req_in, exp_resp);	2✔
152	}
153	});	16✔
154	}
155
156	void Server::RequestBodyHandler(http::IncomingRequestPtr req_in, error::Error err) {	7✔
157	auto maybe_connection = connections_.find(req_in);	7✔
158	if (maybe_connection == connections_.end()) {	7✔
159	// Can happen if the request was cancelled.
160	return;	2✔
161	}
162	auto &connection = maybe_connection->second;	5✔
163
164	if (err != error::NoError) {	5✔
165	connection->logger_.Error("Error while reading incoming request body: " + err.String());	×
166	connections_.erase(req_in);	×
167	return;	×
168	}
169
170	auto exp_resp_out = connection->req_in_->MakeResponse();	5✔
171	if (!exp_resp_out) {	5✔
172	connection->logger_.Error(	×
173	"Could not make outgoing response: " + exp_resp_out.error().String());	×
174	connections_.erase(req_in);	×
175	return;	×
176	}
177	connection->resp_out_ = exp_resp_out.value();	5✔
178	}
179
180	void Server::ResponseHeaderHandler(	7✔
181	http::IncomingRequestPtr req_in, http::ExpectedIncomingResponsePtr exp_resp_in) {
182	auto connection = connections_[req_in];	7✔
183
184	if (!exp_resp_in) {	7✔
185	connection->logger_.Error("Error in incoming response: " + exp_resp_in.error().String());	3✔
186	connections_.erase(req_in);	3✔
187	return;	3✔
188	}
189	connection->resp_in_ = exp_resp_in.value();	4✔
190	auto &resp_in = connection->resp_in_;	4✔
191
192	auto &resp_out = connection->resp_out_;	4✔
193
194	resp_out->SetStatusCodeAndMessage(resp_in->GetStatusCode(), resp_in->GetStatusMessage());	4✔
195	for (auto header : resp_in->GetHeaders()) {	7✔
196	resp_out->SetHeader(header.first, header.second);	3✔
197	}
198
199	auto exp_body_reader = resp_in->MakeBodyAsyncReader();	4✔
200	if (exp_body_reader) {	4✔
201	resp_out->SetAsyncBodyReader(exp_body_reader.value());	3✔
202	} else if (exp_body_reader.error().code != http::MakeError(http::BodyMissingError, "").code) {	1✔
203	connection->logger_.Error(	×
204	"Could not get body reader for response: " + exp_body_reader.error().String());	×
205	connections_.erase(req_in);	×
206	return;	×
207	} // else: if body is missing we don't need to do anything.
208
209	auto &cancelled = cancelled_;	4✔
210	auto err = resp_out->AsyncReply([cancelled, this, req_in](error::Error err) {	9✔
211	if (*cancelled) {	3✔
212	return;	×
213	}
214
215	if (err != error::NoError) {	3✔
216	connections_[req_in]->logger_.Error(	4✔
217	"Error while forwarding response to client: " + err.String());	4✔
218	connections_.erase(req_in);	2✔
219	return;	2✔
220	}
221
222	auto &connection = connections_[req_in];	1✔
223	connection->incoming_request_finished = true;	1✔
224	if (connection->outgoing_request_finished) {	1✔
225	// We are done, remove connection.
226	connections_.erase(req_in);	1✔
227	}
228	});	8✔
229	if (err != error::NoError) {	4✔
230	connection->logger_.Error("Could not forward response to client: " + err.String());	×
231	connections_.erase(req_in);	×
232	return;	×
233	}
234	}
235
236	void Server::ResponseBodyHandler(	2✔
237	http::IncomingRequestPtr req_in, http::ExpectedIncomingResponsePtr exp_resp_in) {
238	auto &connection = connections_[req_in];	2✔
239
240	if (!exp_resp_in) {	2✔
241	connection->logger_.Error(	2✔
242	"Error while reading incoming response body: " + exp_resp_in.error().String());	2✔
243	connections_.erase(req_in);	1✔
244	return;	1✔
245	}
246
247	connection->outgoing_request_finished = true;	1✔
248	if (connection->incoming_request_finished) {	1✔
249	// We are done, remove connection.
250	connections_.erase(req_in);	×
251	}
252	}
253
254	} // namespace http_forwarder
255	} // namespace auth
256	} // namespace mender

mendersoftware / mender / 1018307341

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous