diff options
Diffstat (limited to 'src/components/transport_manager/test/sample_websocket_server.cc')
| -rw-r--r-- | src/components/transport_manager/test/sample_websocket_server.cc | 378 |
1 files changed, 378 insertions, 0 deletions
diff --git a/src/components/transport_manager/test/sample_websocket_server.cc b/src/components/transport_manager/test/sample_websocket_server.cc new file mode 100644 index 0000000000..2cb45242aa --- /dev/null +++ b/src/components/transport_manager/test/sample_websocket_server.cc @@ -0,0 +1,378 @@ +/* + * Copyright (c) 2019, Ford Motor Company + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * Redistributions of source code must retain the above copyright notice, this + * list of conditions and the following disclaimer. + * + * Redistributions in binary form must reproduce the above copyright notice, + * this list of conditions and the following + * disclaimer in the documentation and/or other materials provided with the + * distribution. + * + * Neither the name of the Ford Motor Company nor the names of its contributors + * may be used to endorse or promote products derived from this software + * without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + +#include "transport_manager/cloud/sample_websocket_server.h" + +namespace { +// Report a failure +void Fail(char const* tag, boost::system::error_code ec) { + std::cerr << tag << ": " << ec.message() << "\n"; +} +} // namespace + +namespace sample { +namespace websocket { + +WSSession::WSServer::WSServer(tcp::socket&& socket) + : ws_(std::move(socket)), strand_(ws_.get_executor()) {} + +void WSSession::WSServer::AddURLRoute(const std::string& target) { + url_routes_.insert(ParseRouteFromTarget(target)); +} + +void WSSession::WSServer::Run() { + req_ = {}; + http::async_read( + ws_.next_layer(), + buffer_, + req_, + std::bind(&WSServer::OnWebsocketHandshake, this, std::placeholders::_1)); +} + +void WSSession::WSServer::OnWebsocketHandshake( + const boost::system::error_code& ec) { + if (ec) { + return Fail("ERROR_HTTP_REQUEST_READ", ec); + } + if (websocket::is_upgrade(req_)) { + // Check target + std::string route = ParseRouteFromTarget(req_.target().to_string()); + if (!CanHandleRoute(route)) { + auto error = boost::system::errc::make_error_code( + boost::system::errc::address_not_available); + ws_.next_layer().close(); + return Fail("ERROR_INVALID_TARGET", error); + } + // Accept the websocket handshake + ws_.async_accept( + req_, + boost::asio::bind_executor( + strand_, + std::bind(&WSServer::OnAccept, this, std::placeholders::_1))); + } +} + +void WSSession::WSServer::OnAccept(beast::error_code ec) { + if (ec) { + return Fail("ERROR_WEBSOCKET_HANDSHAKE", ec); + } +} + +bool WSSession::WSServer::CanHandleRoute(const std::string& route) { + if (url_routes_.find(route) == url_routes_.end()) { + return false; + } + return true; +} + +std::string WSSession::WSServer::ParseRouteFromTarget( + const std::string& target) { + std::string route = target; + // Remove fragment + auto fragment_pos = route.find('#'); + if (fragment_pos != std::string::npos) { + route = route.substr(0, fragment_pos); + } + // Remove query + auto query_pos = route.find('?'); + if (query_pos != std::string::npos) { + route = route.substr(0, query_pos); + } + + return route; +} + +WSSession::WSSession(const std::string& address, uint16_t port) + : address_(address) + , port_(port) + , acceptor_(ioc_) + , socket_(ioc_) + , ws_(nullptr) { + endpoint_ = {boost::asio::ip::make_address(address), port}; + boost::system::error_code error; + + // Open the acceptor + acceptor_.open(endpoint_.protocol(), error); + if (error) { + Fail("ERROR_ACCEPTOR_OPEN", error); + return; + } + + // Allow address reuse + acceptor_.set_option(boost::asio::socket_base::reuse_address(true), error); + if (error) { + Fail("ERROR_SET_OPTION", error); + return; + } + + // Bind to the server address + acceptor_.bind(endpoint_, error); + if (error) { + Fail("ERROR_BIND", error); + return; + } + + // Start listening for connections + acceptor_.listen(boost::asio::socket_base::max_listen_connections, error); + if (error) { + Fail("ERROR_LISTEN", error); + return; + } +} + +void WSSession::Run() { + if (acceptor_.is_open()) { + acceptor_.async_accept( + socket_, std::bind(&WSSession::on_accept, this, std::placeholders::_1)); + ioc_.run(); + } +} + +void WSSession::Stop() { + try { + ioc_.stop(); + acceptor_.close(); + } catch (...) { + std::cerr << "Failed to close connection" << std::endl; + } +} + +void WSSession::AddRoute(const std::string& route) { + if (ws_ == nullptr) { + buffered_routes_.push(route); + return; + } + ws_->AddURLRoute(route); +} + +void WSSession::on_accept(boost::system::error_code ec) { + if (ec) { + Fail("ERROR_ON_ACCEPT", ec); + ioc_.stop(); + return; + } + + // Make websocket object and start + ws_ = std::make_shared<WSServer>(std::move(socket_)); + // Load routes stored in buffer + while (!buffered_routes_.empty()) { + ws_->AddURLRoute(buffered_routes_.front()); + buffered_routes_.pop(); + } + ws_->Run(); +} + +WSSSession::WSSServer::WSSServer(tcp::socket&& socket, ssl::context& ctx) + : wss_(std::move(socket), ctx) {} + +void WSSSession::WSSServer::AddURLRoute(const std::string& target) { + url_routes_.insert(ParseRouteFromTarget(target)); +} +void WSSSession::WSSServer::Run() { + // Perform the SSL handshake + wss_.next_layer().async_handshake( + ssl::stream_base::server, + std::bind(&WSSServer::OnSSLHandshake, this, std::placeholders::_1)); +} + +void WSSSession::WSSServer::OnSSLHandshake(beast::error_code ec) { + if (ec) { + return Fail("ERROR_SSL_HANDSHAKE", ec); + } + + req_ = {}; + http::async_read( + wss_.next_layer(), + buffer_, + req_, + std::bind(&WSSServer::OnWebsocketHandshake, this, std::placeholders::_1)); +} + +void WSSSession::WSSServer::OnWebsocketHandshake( + const boost::system::error_code& ec) { + if (ec) { + return Fail("ERROR_HTTP_REQUEST_READ", ec); + } + if (websocket::is_upgrade(req_)) { + // Check target + std::string route = ParseRouteFromTarget(req_.target().to_string()); + if (!CanHandleRoute(route)) { + auto error = boost::system::errc::make_error_code( + boost::system::errc::address_not_available); + wss_.next_layer().next_layer().close(); + return Fail("ERROR_INVALID_TARGET", error); + } + // Accept the websocket handshake + wss_.async_accept( + req_, std::bind(&WSSServer::OnAccept, this, std::placeholders::_1)); + } +} + +void WSSSession::WSSServer::OnAccept(beast::error_code ec) { + if (ec) { + return Fail("ERROR_ON_ACCEPT", ec); + } +} + +bool WSSSession::WSSServer::CanHandleRoute(const std::string& route) { + if (url_routes_.find(route) == url_routes_.end()) { + return false; + } + return true; +} + +std::string WSSSession::WSSServer::ParseRouteFromTarget( + const std::string& target) { + std::string route = target; + // Remove fragment + auto fragment_pos = route.find('#'); + if (fragment_pos != std::string::npos) { + route = route.substr(0, fragment_pos); + } + // Remove query + auto query_pos = route.find('?'); + if (query_pos != std::string::npos) { + route = route.substr(0, query_pos); + } + + return route; +} + +WSSSession::WSSSession(const std::string& address, + uint16_t port, + const std::string& certificate, + const std::string& private_key) + : acceptor_(ioc_) + , socket_(ioc_) + , ctx_(ssl::context::sslv23_server) + , wss_(nullptr) { + beast::error_code ec; + endpoint_ = {boost::asio::ip::make_address(address), port}; + + // Load the certificate + ctx_.use_certificate( + boost::asio::buffer(certificate.c_str(), certificate.size()), + ssl::context::file_format::pem, + ec); + if (ec) { + Fail("ERROR_USE_CERTIFICATE", ec); + return; + } + + // Load the private key + ctx_.use_rsa_private_key( + boost::asio::buffer(private_key.c_str(), private_key.size()), + ssl::context::file_format::pem, + ec); + if (ec) { + Fail("ERROR_USE_RSA_PRIVATE_KEY", ec); + return; + } + + // Open the acceptor + acceptor_.open(endpoint_.protocol(), ec); + if (ec) { + Fail("EEROR_ACCEPTOR_OPEN", ec); + return; + } + + // Allow address reuse + acceptor_.set_option(net::socket_base::reuse_address(true), ec); + if (ec) { + Fail("ERROR_SET_OPTION", ec); + return; + } + + // Bind to the server address + acceptor_.bind(endpoint_, ec); + if (ec) { + Fail("ERROR_BIND", ec); + return; + } + + // Start listening for connections + acceptor_.listen(net::socket_base::max_listen_connections, ec); + if (ec) { + Fail("ERROR_LISTEN", ec); + return; + } +} + +// Start accepting incoming connections +void WSSSession::Run() { + do_accept(); +} + +void WSSSession::Stop() { + try { + ioc_.stop(); + acceptor_.close(); + } catch (...) { + std::cerr << "Failed to close connection" << std::endl; + } +} + +void WSSSession::AddRoute(const std::string& route) { + if (wss_ == nullptr) { + buffered_routes_.push(route); + return; + } + wss_->AddURLRoute(route); +} + +void WSSSession::do_accept() { + if (acceptor_.is_open()) { + acceptor_.async_accept( + socket_, + std::bind(&WSSSession::on_accept, this, std::placeholders::_1)); + ioc_.run(); + } +} + +void WSSSession::on_accept(boost::system::error_code ec) { + if (ec) { + Fail("ERROR_ON_ACCEPT", ec); + ioc_.stop(); + return; + } + // Create the session and run it + wss_ = std::make_shared<WSSServer>(std::move(socket_), ctx_); + // Load routes stored in buffer + while (!buffered_routes_.empty()) { + wss_->AddURLRoute(buffered_routes_.front()); + buffered_routes_.pop(); + } + wss_->Run(); +} + +} // namespace websocket +} // namespace sample
\ No newline at end of file |