// Copyright (c) 2014-2023, The Monero Project // // All rights reserved. // // Redistribution and use in source and binary forms, with or without modification, are // permitted provided that the following conditions are met: // // 1. Redistributions of source code must retain the above copyright notice, this list of // conditions and the following disclaimer. // // 2. 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. // // 3. Neither the name of the copyright holder 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. // // Parts of this file are originally copyright (c) 2012-2013 The Cryptonote developers #include #include #include #include #include #include "gtest/gtest.h" #include "include_base_utils.h" #include "string_tools.h" #include "net/abstract_tcp_server2.h" #include "net/levin_protocol_handler_async.h" namespace { const uint32_t test_server_port = 5626; const std::string test_server_host("127.0.0.1"); struct test_connection_context : public epee::net_utils::connection_context_base { }; struct test_protocol_handler_config { }; struct test_protocol_handler { typedef test_connection_context connection_context; typedef test_protocol_handler_config config_type; test_protocol_handler(epee::net_utils::i_service_endpoint* /*psnd_hndlr*/, config_type& /*config*/, connection_context& /*conn_context*/) { } void after_init_connection() { } void handle_qued_callback() { } bool release_protocol() { return true; } bool handle_recv(const void* /*data*/, size_t /*size*/) { return false; } }; typedef epee::net_utils::boosted_tcp_server test_tcp_server; } TEST(boosted_tcp_server, worker_threads_are_exception_resistant) { test_tcp_server srv(epee::net_utils::e_connection_type_RPC); // RPC disables network limit for unit tests ASSERT_TRUE(srv.init_server(test_server_port, test_server_host)); boost::mutex mtx; boost::condition_variable cond; int counter = 0; auto counter_incrementer = [&counter, &cond, &mtx]() { boost::unique_lock lock(mtx); ++counter; if (4 <= counter) { cond.notify_one(); } }; // 2 theads, but 4 exceptions ASSERT_TRUE(srv.run_server(2, false)); ASSERT_TRUE(srv.async_call([&counter_incrementer]() { counter_incrementer(); throw std::runtime_error("test 1"); })); ASSERT_TRUE(srv.async_call([&counter_incrementer]() { counter_incrementer(); throw std::string("test 2"); })); ASSERT_TRUE(srv.async_call([&counter_incrementer]() { counter_incrementer(); throw "test 3"; })); ASSERT_TRUE(srv.async_call([&counter_incrementer]() { counter_incrementer(); throw 4; })); { boost::unique_lock lock(mtx); ASSERT_TRUE(cond.wait_for(lock, boost::chrono::seconds(5), [&counter]{ return counter == 4; })); } // Check if threads are alive counter = 0; //auto counter_incrementer = [&counter]() { counter.fetch_add(1); epee::misc_utils::sleep_no_w(counter.load() * 10); }; ASSERT_TRUE(srv.async_call(counter_incrementer)); ASSERT_TRUE(srv.async_call(counter_incrementer)); ASSERT_TRUE(srv.async_call(counter_incrementer)); ASSERT_TRUE(srv.async_call(counter_incrementer)); { boost::unique_lock lock(mtx); ASSERT_TRUE(cond.wait_for(lock, boost::chrono::seconds(5), [&counter]{ return counter == 4; })); } srv.send_stop_signal(); ASSERT_TRUE(srv.timed_wait_server_stop(5 * 1000)); ASSERT_TRUE(srv.deinit_server()); } TEST(test_epee_connection, test_lifetime) { struct context_t: epee::net_utils::connection_context_base { static constexpr size_t get_max_bytes(int) noexcept { return -1; } static constexpr int handshake_command() noexcept { return 1001; } static constexpr bool handshake_complete() noexcept { return true; } }; using functional_obj_t = std::function; struct command_handler_t: epee::levin::levin_commands_handler { size_t delay; functional_obj_t on_connection_close_f; command_handler_t(size_t delay = 0, functional_obj_t on_connection_close_f = nullptr ): delay(delay), on_connection_close_f(on_connection_close_f) {} virtual int invoke(int, const epee::span, epee::byte_stream&, context_t&) override { epee::misc_utils::sleep_no_w(delay); return {}; } virtual int notify(int, const epee::span, context_t&) override { return {}; } virtual void callback(context_t&) override {} virtual void on_connection_new(context_t&) override {} virtual void on_connection_close(context_t&) override { if (on_connection_close_f) on_connection_close_f(); } virtual ~command_handler_t() override {} static void destroy(epee::levin::levin_commands_handler* ptr) { delete ptr; } }; using handler_t = epee::levin::async_protocol_handler; using connection_t = epee::net_utils::connection; using connection_ptr = boost::shared_ptr; using shared_state_t = typename connection_t::shared_state; using shared_state_ptr = std::shared_ptr; using shared_states_t = std::vector; using tag_t = boost::uuids::uuid; using tags_t = std::vector; using io_context_t = boost::asio::io_service; using endpoint_t = boost::asio::ip::tcp::endpoint; using work_t = boost::asio::io_service::work; using work_ptr = std::shared_ptr; using workers_t = std::vector; using server_t = epee::net_utils::boosted_tcp_server; using lock_t = std::mutex; using lock_guard_t = std::lock_guard; using connection_weak_ptr = boost::weak_ptr; struct shared_conn_t { lock_t lock; connection_weak_ptr conn; }; using shared_conn_ptr = std::shared_ptr; io_context_t io_context; work_ptr work(std::make_shared(io_context)); workers_t workers; while (workers.size() < 4) { workers.emplace_back([&io_context]{ io_context.run(); }); } endpoint_t endpoint(boost::asio::ip::address::from_string("127.0.0.1"), 5262); server_t server(epee::net_utils::e_connection_type_P2P); server.init_server(endpoint.port(), endpoint.address().to_string(), 0, "", false, true, epee::net_utils::ssl_support_t::e_ssl_support_disabled ); server.run_server(2, false); server.get_config_shared()->set_handler(new command_handler_t, &command_handler_t::destroy); io_context.post([&io_context, &work, &endpoint, &server]{ shared_state_ptr shared_state; auto scope_exit_handler = epee::misc_utils::create_scope_leave_handler([&work, &shared_state]{ work.reset(); if (shared_state) shared_state->set_handler(nullptr, nullptr); }); shared_state = std::make_shared(); shared_state->set_handler(new command_handler_t, &command_handler_t::destroy); auto create_connection = [&io_context, &endpoint, &shared_state] { connection_ptr conn(new connection_t(io_context, shared_state, {}, {})); conn->socket().connect(endpoint); conn->start({}, {}); context_t context; conn->get_context(context); auto tag = context.m_connection_id; return tag; }; ASSERT_TRUE(shared_state->get_connections_count() == 0); auto tag = create_connection(); ASSERT_TRUE(shared_state->get_connections_count() == 1); bool success = shared_state->for_connection(tag, [shared_state](context_t& context){ shared_state->close(context.m_connection_id); context.m_remote_address.get_zone(); return true; }); ASSERT_TRUE(success); ASSERT_TRUE(shared_state->get_connections_count() == 0); constexpr auto N = 8; tags_t tags(N); for(auto &t: tags) t = create_connection(); ASSERT_TRUE(shared_state->get_connections_count() == N); size_t index = 0; success = shared_state->foreach_connection([&index, shared_state, &tags, &create_connection](context_t& context){ if (!index) for (const auto &t: tags) shared_state->close(t); shared_state->close(context.m_connection_id); context.m_remote_address.get_zone(); ++index; for(auto i = 0; i < N; ++i) create_connection(); return true; }); ASSERT_TRUE(success); ASSERT_TRUE(index == N); ASSERT_TRUE(shared_state->get_connections_count() == N * N); index = 0; success = shared_state->foreach_connection([&index, shared_state](context_t& context){ shared_state->close(context.m_connection_id); context.m_remote_address.get_zone(); ++index; return true; }); ASSERT_TRUE(success); ASSERT_TRUE(index == N * N); ASSERT_TRUE(shared_state->get_connections_count() == 0); while (shared_state->sock_count); ASSERT_TRUE(shared_state->get_connections_count() == 0); constexpr auto DELAY = 30; constexpr auto TIMEOUT = 1; while (server.get_connections_count()) { server.get_config_shared()->del_in_connections( server.get_config_shared()->get_in_connections_count() ); } server.get_config_shared()->set_handler(new command_handler_t(DELAY), &command_handler_t::destroy); for (auto i = 0; i < N; ++i) { tag = create_connection(); ASSERT_TRUE(shared_state->get_connections_count() == 1); success = shared_state->invoke_async(1, epee::levin::message_writer{}, tag, [](int, const epee::span, context_t&){}, TIMEOUT); ASSERT_TRUE(success); while (shared_state->sock_count == 1) { success = shared_state->foreach_connection([&shared_state, &tag](context_t&){ return shared_state->request_callback(tag); }); ASSERT_TRUE(success); } shared_state->close(tag); ASSERT_TRUE(shared_state->get_connections_count() == 0); } while (shared_state->sock_count); constexpr auto ZERO_DELAY = 0; size_t counter = 0; shared_state->set_handler(new command_handler_t(ZERO_DELAY, [&counter]{ ASSERT_TRUE(counter++ == 0); } ), &command_handler_t::destroy ); connection_ptr conn(new connection_t(io_context, shared_state, {}, {})); conn->socket().connect(endpoint); conn->start({}, {}); ASSERT_TRUE(shared_state->get_connections_count() == 1); shared_state->del_out_connections(1); ASSERT_TRUE(shared_state->get_connections_count() == 0); conn.reset(); while (shared_state->sock_count); shared_conn_ptr shared_conn(std::make_shared()); shared_state->set_handler(new command_handler_t(ZERO_DELAY, [shared_state, shared_conn]{ { connection_ptr conn; { lock_guard_t guard(shared_conn->lock); conn = shared_conn->conn.lock(); shared_conn->conn.reset(); } if (conn) conn->cancel(); } const auto success = shared_state->foreach_connection([](context_t&){ return true; }); ASSERT_TRUE(success); } ), &command_handler_t::destroy ); for (auto i = 0; i < N * N * N; ++i) { { connection_ptr conn(new connection_t(io_context, shared_state, {}, {})); conn->socket().connect(endpoint); conn->start({}, {}); lock_guard_t guard(shared_conn->lock); shared_conn->conn = conn; } ASSERT_TRUE(shared_state->get_connections_count() == 1); shared_state->del_out_connections(1); while (shared_state->sock_count); ASSERT_TRUE(shared_state->get_connections_count() == 0); } shared_states_t shared_states; while (shared_states.size() < 2) { shared_states.emplace_back(std::make_shared()); shared_states.back()->set_handler(new command_handler_t(ZERO_DELAY, [&shared_states]{ for (auto &s: shared_states) { auto success = s->foreach_connection([](context_t&){ return true; }); ASSERT_TRUE(success); } } ), &command_handler_t::destroy ); } workers_t workers; for (auto &s: shared_states) { workers.emplace_back([&io_context, &s, &endpoint]{ for (auto i = 0; i < N * N; ++i) { connection_ptr conn(new connection_t(io_context, s, {}, {})); conn->socket().connect(endpoint); conn->start({}, {}); io_context.post([conn]{ conn->cancel(); }); conn.reset(); s->del_out_connections(1); while (s->sock_count); } }); } for (;workers.size(); workers.pop_back()) workers.back().join(); for (auto &s: shared_states) { workers.emplace_back([&io_context, &s, &endpoint]{ for (auto i = 0; i < N * N; ++i) { connection_ptr conn(new connection_t(io_context, s, {}, {})); conn->socket().connect(endpoint); conn->start({}, {}); conn->cancel(); while (conn.use_count() > 1); s->foreach_connection([&io_context, &s, &endpoint, &conn](context_t& context){ conn.reset(new connection_t(io_context, s, {}, {})); conn->socket().connect(endpoint); conn->start({}, {}); conn->cancel(); while (conn.use_count() > 1); conn.reset(); return true; }); while (s->sock_count); } }); } for (;workers.size(); workers.pop_back()) workers.back().join(); for (auto &s: shared_states) { workers.emplace_back([&io_context, &s, &endpoint]{ for (auto i = 0; i < N; ++i) { connection_ptr conn(new connection_t(io_context, s, {}, {})); conn->socket().connect(endpoint); conn->start({}, {}); context_t context; conn->get_context(context); auto tag = context.m_connection_id; conn->cancel(); while (conn.use_count() > 1); s->for_connection(tag, [&io_context, &s, &endpoint, &conn](context_t& context){ conn.reset(new connection_t(io_context, s, {}, {})); conn->socket().connect(endpoint); conn->start({}, {}); conn->cancel(); while (conn.use_count() > 1); conn.reset(); return true; }); while (s->sock_count); } }); } for (;workers.size(); workers.pop_back()) workers.back().join(); for (auto &s: shared_states) { workers.emplace_back([&io_context, &s, &endpoint]{ for (auto i = 0; i < N; ++i) { connection_ptr conn(new connection_t(io_context, s, {}, {})); conn->socket().connect(endpoint); conn->start({}, {}); context_t context; conn->get_context(context); auto tag = context.m_connection_id; io_context.post([conn]{ conn->cancel(); }); conn.reset(); s->close(tag); while (s->sock_count); } }); } for (;workers.size(); workers.pop_back()) workers.back().join(); while (server.get_connections_count()) { server.get_config_shared()->del_in_connections( server.get_config_shared()->get_in_connections_count() ); } }); for (auto& w: workers) { w.join(); } server.send_stop_signal(); server.timed_wait_server_stop(5 * 1000); server.deinit_server(); } TEST(test_epee_connection, ssl_shutdown) { struct context_t: epee::net_utils::connection_context_base { static constexpr size_t get_max_bytes(int) noexcept { return -1; } static constexpr int handshake_command() noexcept { return 1001; } static constexpr bool handshake_complete() noexcept { return true; } }; struct command_handler_t: epee::levin::levin_commands_handler { virtual int invoke(int, const epee::span, epee::byte_stream&, context_t&) override { return {}; } virtual int notify(int, const epee::span, context_t&) override { return {}; } virtual void callback(context_t&) override {} virtual void on_connection_new(context_t&) override {} virtual void on_connection_close(context_t&) override { } virtual ~command_handler_t() override {} static void destroy(epee::levin::levin_commands_handler* ptr) { delete ptr; } }; using handler_t = epee::levin::async_protocol_handler; using io_context_t = boost::asio::io_service; using endpoint_t = boost::asio::ip::tcp::endpoint; using server_t = epee::net_utils::boosted_tcp_server; using socket_t = boost::asio::ip::tcp::socket; using ssl_socket_t = boost::asio::ssl::stream; using ssl_context_t = boost::asio::ssl::context; using ec_t = boost::system::error_code; endpoint_t endpoint(boost::asio::ip::address::from_string("127.0.0.1"), 5263); server_t server(epee::net_utils::e_connection_type_P2P); server.init_server(endpoint.port(), endpoint.address().to_string(), 0, "", false, true, epee::net_utils::ssl_support_t::e_ssl_support_enabled ); server.get_config_shared()->set_handler(new command_handler_t, &command_handler_t::destroy); server.run_server(2, false); ssl_context_t ssl_context{boost::asio::ssl::context::sslv23}; io_context_t io_context; ssl_socket_t socket(io_context, ssl_context); ec_t ec; socket.next_layer().connect(endpoint, ec); EXPECT_EQ(ec.value(), 0); socket.handshake(boost::asio::ssl::stream_base::client, ec); EXPECT_EQ(ec.value(), 0); std::this_thread::sleep_for(std::chrono::milliseconds(100)); while (server.get_config_shared()->get_connections_count() < 1); server.get_config_shared()->del_in_connections(1); while (server.get_config_shared()->get_connections_count() > 0); server.send_stop_signal(); EXPECT_TRUE(server.timed_wait_server_stop(5 * 1000)); server.deinit_server(); socket.next_layer().shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec); socket.next_layer().close(ec); socket.shutdown(ec); } TEST(test_epee_connection, ssl_handshake) { using io_context_t = boost::asio::io_service; using work_t = boost::asio::io_service::work; using work_ptr = std::shared_ptr; using workers_t = std::vector; using socket_t = boost::asio::ip::tcp::socket; using ssl_socket_t = boost::asio::ssl::stream; using ssl_socket_ptr = std::unique_ptr; using ssl_options_t = epee::net_utils::ssl_options_t; io_context_t io_context; work_ptr work(std::make_shared(io_context)); workers_t workers; auto constexpr N = 2; while (workers.size() < N) { workers.emplace_back([&io_context]{ io_context.run(); }); } ssl_options_t ssl_options{{}}; auto ssl_context = ssl_options.create_context(); for (size_t i = 0; i < N * N * N; ++i) { ssl_socket_ptr ssl_socket(new ssl_socket_t(io_context, ssl_context)); ssl_socket->next_layer().open(boost::asio::ip::tcp::v4()); for (size_t i = 0; i < N; ++i) { io_context.post([]{ std::this_thread::sleep_for(std::chrono::milliseconds(50)); }); } EXPECT_EQ( ssl_options.handshake( *ssl_socket, ssl_socket_t::server, {}, {}, std::chrono::milliseconds(0) ), false ); ssl_socket->next_layer().close(); ssl_socket.reset(); } work.reset(); for (;workers.size(); workers.pop_back()) workers.back().join(); } TEST(boosted_tcp_server, strand_deadlock) { using context_t = epee::net_utils::connection_context_base; using lock_t = std::mutex; using unique_lock_t = std::unique_lock; struct config_t { using condition_t = std::condition_variable_any; using lock_guard_t = std::lock_guard; void notify_success() { lock_guard_t guard(lock); success = true; condition.notify_all(); } lock_t lock; condition_t condition; bool success; }; struct handler_t { using config_type = config_t; using connection_context = context_t; using byte_slice_t = epee::byte_slice; using socket_t = epee::net_utils::i_service_endpoint; handler_t(socket_t *socket, config_t &config, context_t &context): socket(socket), config(config), context(context) {} void after_init_connection() { unique_lock_t guard(lock); if (!context.m_is_income) { guard.unlock(); socket->do_send(byte_slice_t{"."}); } } void handle_qued_callback() { } bool handle_recv(const char *data, size_t bytes_transferred) { unique_lock_t guard(lock); if (!context.m_is_income) { if (context.m_recv_cnt == 1024) { guard.unlock(); socket->do_send(byte_slice_t{"."}); } } else { if (context.m_recv_cnt == 1) { for(size_t i = 0; i < 1024; ++i) { guard.unlock(); socket->do_send(byte_slice_t{"."}); guard.lock(); } } else if(context.m_recv_cnt == 2) { guard.unlock(); socket->close(); } } return true; } void release_protocol() { unique_lock_t guard(lock); if(!context.m_is_income && context.m_recv_cnt == 1024 && context.m_send_cnt == 2 ) { guard.unlock(); config.notify_success(); } } lock_t lock; socket_t *socket; config_t &config; context_t &context; }; using server_t = epee::net_utils::boosted_tcp_server; using endpoint_t = boost::asio::ip::tcp::endpoint; endpoint_t endpoint(boost::asio::ip::address::from_string("127.0.0.1"), 5262); server_t server(epee::net_utils::e_connection_type_P2P); server.init_server( endpoint.port(), endpoint.address().to_string(), {}, {}, {}, true, epee::net_utils::ssl_support_t::e_ssl_support_disabled ); server.run_server(2, {}); server.async_call( [&]{ context_t context; ASSERT_TRUE( server.connect( endpoint.address().to_string(), std::to_string(endpoint.port()), 5, context, "0.0.0.0", epee::net_utils::ssl_support_t::e_ssl_support_disabled ) ); } ); { unique_lock_t guard(server.get_config_object().lock); EXPECT_TRUE( server.get_config_object().condition.wait_for( guard, std::chrono::seconds(5), [&] { return server.get_config_object().success; } ) ); } server.send_stop_signal(); server.timed_wait_server_stop(5 * 1000); server.deinit_server(); }