diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..0ff2df7 --- /dev/null +++ b/TODO.md @@ -0,0 +1,16 @@ +# Follow-up work + +## C++ TLS and asynchronous lifecycle hardening + +The focused multithreaded TLS half-close test passes with the quality, AddressSanitizer, and ThreadSanitizer builds as of 2026-08-13. The demonstrated memory-safety defect was an allocator-lifetime issue at type-erased asynchronous-operation boundaries, not an OpenSSL or TLS defect. The affected boundaries have dedicated fixes and regression tests in the current worktree. + +This area is deliberately not the critical path of the current guide and sample validation. Before declaring the TLS lifecycle audit exhaustive, complete the following follow-up matrix: + +- Diagnose the reproducible Windows shared-library timeout in `check_tls_half_close_preserves_receive_direction`: both the initial CI run and its isolated rerun reached the 10-second deadline at `test_io_common_stream_tls.cpp:861` and then aborted with `0xc0000409`, while the Windows static build and every Linux/macOS variant passed. Replace the terminal assertion with operation-level timeout evidence before changing behavior, identify which half-close completion is missing, and prove the correction through repeated Windows shared-library runs. Do not classify this as an OpenSSL defect without that evidence. +- Exercise TLS 1.2 and TLS 1.3 handshake, cancellation, half-close, peer-close, timeout, and abrupt-reset paths with one and several `io_context` worker threads. +- Repeat client/server interoperability in both directions for Go and C++, including concurrent streams and shutdown during backpressure. +- Run long-duration and high-concurrency stress tests under AddressSanitizer and ThreadSanitizer, and retain machine-readable evidence in CI. +- Add the corresponding stateful associated-allocator abandonment checks to every remaining type-erased operation boundary, including Windows-only handles in Windows CI. +- Remove the stale OpenSSL 1.1 linker search path emitted by macOS builds after verifying that packaging remains compatible with the supported OpenSSL versions. + +Completion requires zero sanitizer findings, deterministic cancellation, exactly-once completion, no handler after owner destruction, bounded shutdown time, and no measurable throughput or latency regression against the recorded baseline. diff --git a/bin/ncat/bin/main.cpp b/bin/ncat/bin/main.cpp index d18a1f6..b5926da 100644 --- a/bin/ncat/bin/main.cpp +++ b/bin/ncat/bin/main.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include @@ -48,9 +49,26 @@ this program is distributed with the rstream C++ tools. See https://rstream.io/d const auto version = std::string("rstream-ncat ") + RSTREAM_VERSION; +static std::vector normalize_cli_arguments(int argc, char** argv) +{ + std::vector result; + for (int index = 1; index < argc; ++index) { + std::string argument = argv[index]; + if (argument.starts_with("-c=") || argument.starts_with("-e=")) { + result.emplace_back(argument.substr(0, 2)); + result.emplace_back(argument.substr(3)); + } + else { + result.emplace_back(std::move(argument)); + } + } + return result; +} + int run(int argc, char** argv) { - auto args = docopt::docopt(USAGE, {argv + 1, argv + argc}, true, version); + auto cli_args = normalize_cli_arguments(argc, argv); + auto args = docopt::docopt(USAGE, cli_args, true, version); bool verbose = false; { auto it = args.find("--verbose"); diff --git a/bin/ncat/lib/rstream/ncat/client.cpp b/bin/ncat/lib/rstream/ncat/client.cpp index 9e76fc5..ff725f9 100644 --- a/bin/ncat/lib/rstream/ncat/client.cpp +++ b/bin/ncat/lib/rstream/ncat/client.cpp @@ -455,9 +455,18 @@ void client::impl::on_read_std_in(const boost::system::error_code& error_code, s } else if (eos) { m_std_in_eos = true; -#ifndef RSTREAM_WITH_IO_STREAMS - boost::system::error_code tmp; - m_socket.shutdown(boost::asio::socket_base::shutdown_send, tmp); +#ifdef RSTREAM_WITH_IO_STREAMS + m_socket.async_shutdown_send(boost::asio::bind_executor(m_strand, [self = shared_from_this()](const boost::system::error_code& shutdown_error) { + if (self->m_state == state::connected && shutdown_error && !core::helpers::is_eof_error(shutdown_error)) { + self->on_error(shutdown_error); + } + })); +#else + boost::system::error_code shutdown_error; + m_socket.shutdown(boost::asio::socket_base::shutdown_send, shutdown_error); + if (shutdown_error) { + on_error(shutdown_error); + } #endif } else { diff --git a/bin/ncat/lib/rstream/ncat/server.cpp b/bin/ncat/lib/rstream/ncat/server.cpp index 478bcf7..22b6e86 100644 --- a/bin/ncat/lib/rstream/ncat/server.cpp +++ b/bin/ncat/lib/rstream/ncat/server.cpp @@ -44,6 +44,13 @@ #include #include #endif +#ifndef _WIN32 +#if __has_include() +#include +#else +#include +#endif +#endif #include #include @@ -52,6 +59,8 @@ #ifdef _WIN32 #include #include +#else +#include #endif // clang-format on @@ -64,6 +73,9 @@ #include #include #include +#ifndef _WIN32 +#include +#endif #include #ifdef RSTREAM_WITH_IO_STREAMS #include @@ -389,7 +401,7 @@ class RSTREAM_GNUC_INTERNAL server::impl::session_proxy : public session, public class RSTREAM_GNUC_INTERNAL server::impl::session_exec : public session, public std::enable_shared_from_this { public: - session_exec(socket_type&& downstream_socket, const settings_server& settings, const session_id_type& session_id, const exec& exec, bool downstream_half_close); + session_exec(socket_type&& downstream_socket, const settings_server& settings, const session_id_type& session_id, const exec& exec); void async_run(async_run_completion_handler&& handler) override; @@ -454,15 +466,17 @@ class RSTREAM_GNUC_INTERNAL server::impl::session_exec : public session, public const exec m_exec; - const bool m_downstream_half_close; - core::logger m_logger; state m_state; async_run_completion_handler m_handler; +#ifdef _WIN32 boost::process::async_pipe m_child_stdin; +#else + rstream::core::posix::child_stdin m_child_stdin; +#endif boost::process::async_pipe m_child_stdout; @@ -726,7 +740,7 @@ void server::impl::on_accept(const boost::system::error_code& error_code) session_ptr = std::make_shared(std::move(m_socket), m_settings, session_id, boost::get(m_config.m_remote)); } else if (m_config.m_remote.type() == typeid(exec)) { - session_ptr = std::make_shared(std::move(m_socket), m_settings, session_id, boost::get(m_config.m_remote), m_config.m_local.m_url.scheme() == "tcp"); + session_ptr = std::make_shared(std::move(m_socket), m_settings, session_id, boost::get(m_config.m_remote)); } if (session_ptr) { m_sessions.insert(std::make_pair(session_id, session_ptr)); @@ -1172,17 +1186,20 @@ void server::impl::session_proxy::on_close(const boost::system::error_code& erro m_buffer_read_upstream = nullptr; } -server::impl::session_exec::session_exec(socket_type&& downstream_socket, const settings_server& settings, const session_id_type& session_id, const exec& exec, bool downstream_half_close) +server::impl::session_exec::session_exec(socket_type&& downstream_socket, const settings_server& settings, const session_id_type& session_id, const exec& exec) : m_executor(downstream_socket.get_executor()), m_strand(m_executor), m_settings(settings), m_downstream_socket(std::move(downstream_socket)), m_session_id(session_id), m_exec(exec), - m_downstream_half_close(downstream_half_close), m_logger({"rstream", "ncat", "session", fmt::format("#{}", session_id)}), m_state(state::null), +#ifdef _WIN32 m_child_stdin(get_io_context(m_executor)), +#else + m_child_stdin(m_executor), +#endif m_child_stdout(get_io_context(m_executor)), m_child_stderr(get_io_context(m_executor)), m_child_stdout_eos(false), @@ -1287,9 +1304,14 @@ void server::impl::session_exec::start_child() m_logger->trace("starting child process [shell: {} | cmd: {}]", shell, m_exec.m_cmd); m_child = std::make_shared(shell, boost::process::args(args), +#ifdef _WIN32 boost::process::std_in m_child_stdout, +#else + boost::process::posix::fd.bind(STDIN_FILENO, m_child_stdin.child_native_handle()), + boost::process::std_out > m_child_stdout, +#endif boost::process::std_err > m_child_stderr, boost::process::on_exit = completion_handler, get_io_context(m_executor)); @@ -1317,9 +1339,14 @@ void server::impl::session_exec::start_child() m_logger->trace("starting child process [exe: {} | args: {}]", exe, args_stream.str()); m_child = std::make_shared(exe, boost::process::args(args), +#ifdef _WIN32 boost::process::std_in m_child_stdout, +#else + boost::process::posix::fd.bind(STDIN_FILENO, m_child_stdin.child_native_handle()), + boost::process::std_out > m_child_stdout, +#endif boost::process::std_err > m_child_stderr, boost::process::on_exit = completion_handler, get_io_context(m_executor)); @@ -1329,6 +1356,9 @@ void server::impl::session_exec::start_child() catch (...) { exception_ptr = std::current_exception(); } +#ifndef _WIN32 + m_child_stdin.close_child_end(); +#endif if (exception_ptr) { try { std::rethrow_exception(exception_ptr); @@ -1409,10 +1439,6 @@ void server::impl::session_exec::on_read_downstream(const boost::system::error_c } if (error_code) { if (core::helpers::is_eof_error(error_code)) { - if (!m_downstream_half_close) { - on_close(boost::system::error_code()); - return; - } { boost::system::error_code tmp; m_child_stdin.close(tmp); @@ -1444,7 +1470,11 @@ void server::impl::session_exec::do_write_child() auto completion_handler = [self = shared_from_this(), buffer](const boost::system::error_code& error_code, std::size_t size) { self->on_write_child(error_code, size); }; +#ifdef _WIN32 boost::asio::async_write(m_child_stdin, core::helpers::const_memory_sequence(*buffer), boost::asio::bind_executor(m_strand, std::move(completion_handler))); +#else + boost::asio::async_write(m_child_stdin.stream(), core::helpers::const_memory_sequence(*buffer), boost::asio::bind_executor(m_strand, std::move(completion_handler))); +#endif } void server::impl::session_exec::on_write_child(const boost::system::error_code& error_code, std::size_t size) diff --git a/bin/tunnel/lib/rstream/tunnel/proxy.cpp b/bin/tunnel/lib/rstream/tunnel/proxy.cpp index bc0400e..9671662 100644 --- a/bin/tunnel/lib/rstream/tunnel/proxy.cpp +++ b/bin/tunnel/lib/rstream/tunnel/proxy.cpp @@ -189,6 +189,10 @@ class RSTREAM_GNUC_INTERNAL proxy::impl::session : public std::enable_shared_fro void on_read(const boost::system::error_code& error_code, std::size_t size, type type); + void do_shutdown_send(type type); + + void on_shutdown_send(const boost::system::error_code& error_code); + void do_write(type type); void on_write(const boost::system::error_code& error_code, std::size_t size, type type); @@ -221,6 +225,10 @@ class RSTREAM_GNUC_INTERNAL proxy::impl::session : public std::enable_shared_fro state m_state; + bool m_downstream_read_closed; + + bool m_upstream_read_closed; + async_run_completion_handler m_handler; std::shared_ptr m_buffer_read_downstream; @@ -574,7 +582,9 @@ proxy::impl::session::session(downstream_socket_type&& downstream_socket, const m_session_id(session_id), m_upstream_address(upstream_address), m_logger({"rstream", "tunnel", "session", fmt::format("#{}", session_id)}), - m_state(state::null) + m_state(state::null), + m_downstream_read_closed(false), + m_upstream_read_closed(false) { std::stringstream str; { @@ -790,14 +800,71 @@ void proxy::impl::session::on_read(const boost::system::error_code& error_code, if (m_state != state::connected) { return; } - if (error_code) { + const auto eof = error_code && core::helpers::is_eof_error(error_code); + if (error_code && !eof) { on_error(error_code); + return; } - else { + if (eof) { + if (type == type::downstream) { + m_downstream_read_closed = true; + } + else { + m_upstream_read_closed = true; + } + } + if (size > 0) { auto& buffer = type == type::downstream ? *m_buffer_read_downstream : *m_buffer_read_upstream; buffer.set_size(size); do_write(type == type::downstream ? type::upstream : type::downstream); } + else if (eof) { + do_shutdown_send(type == type::downstream ? type::upstream : type::downstream); + } + else { + do_read(type); + } +} + +void proxy::impl::session::do_shutdown_send(type type) +{ +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif + if (type == type::downstream) { + auto completion_handler = boost::asio::bind_executor( + m_strand, + [ptr = shared_from_this()](const boost::system::error_code& error_code) { ptr->on_shutdown_send(error_code); }); + m_downstream_socket.async_shutdown_send(std::move(completion_handler)); + } + else { +#ifdef RSTREAM_WITH_IO_STREAMS + auto completion_handler = boost::asio::bind_executor( + m_strand, + [ptr = shared_from_this()](const boost::system::error_code& error_code) { ptr->on_shutdown_send(error_code); }); + m_upstream_socket.async_shutdown_send(std::move(completion_handler)); +#else + boost::system::error_code error_code; + m_upstream_socket.shutdown(boost::asio::socket_base::shutdown_send, error_code); + on_shutdown_send(error_code); +#endif + } +} + +void proxy::impl::session::on_shutdown_send(const boost::system::error_code& error_code) +{ +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif + if (m_state != state::connected) { + return; + } + if (error_code && !core::helpers::is_eof_error(error_code)) { + on_error(error_code); + } + else if (m_downstream_read_closed && m_upstream_read_closed) { + on_close(boost::system::error_code()); + } } void proxy::impl::session::do_write(type type) @@ -831,7 +898,13 @@ void proxy::impl::session::on_write(const boost::system::error_code& error_code, on_error(error_code); } else { - do_read(type == type::downstream ? type::upstream : type::downstream); + const auto source_closed = type == type::downstream ? m_upstream_read_closed : m_downstream_read_closed; + if (source_closed) { + do_shutdown_send(type); + } + else { + do_read(type == type::downstream ? type::upstream : type::downstream); + } } } diff --git a/bin/webtty/lib/rstream/webtty/client.cpp b/bin/webtty/lib/rstream/webtty/client.cpp index 42fc08b..04bc4e6 100644 --- a/bin/webtty/lib/rstream/webtty/client.cpp +++ b/bin/webtty/lib/rstream/webtty/client.cpp @@ -1195,12 +1195,18 @@ void client::impl::on_send_message(const std::error_code& error_code, enum loop if (m_state == state::null || m_state == state::disconnected) { return; } - if (error_code) { - on_error(error_code); - } - else if (m_remote_return_code) { + if (m_remote_return_code) { finish_cmd_if_idle(); } + else if (error_code) { + if (m_state == state::connected) { + set_state(state::disconnecting); + arm_state_timer(m_settings.m_common.m_timeouts_ms.m_close); + } + if (!m_error_code) { + m_error_code = error_code; + } + } else { switch (loop) { case loop::read_std_in: diff --git a/bin/webtty/lib/rstream/webtty/detail/process.hpp b/bin/webtty/lib/rstream/webtty/detail/process.hpp index 2a83505..7876f7e 100644 --- a/bin/webtty/lib/rstream/webtty/detail/process.hpp +++ b/bin/webtty/lib/rstream/webtty/detail/process.hpp @@ -29,6 +29,13 @@ #else #include #endif +#ifndef _WIN32 +#if __has_include() +#include +#else +#include +#endif +#endif #include #include @@ -109,10 +116,24 @@ std::shared_ptr make_child(stream::ptr stream_ptr, Args&& } else { auto stream_ptr_pipe = std::dynamic_pointer_cast(stream_ptr); - child = std::make_shared(boost::process::std_out > stream_ptr_pipe->stream(stream::type::std_out), - boost::process::std_err > stream_ptr_pipe->stream(stream::type::std_err), - boost::process::std_in < stream_ptr_pipe->stream(stream::type::std_in), - std::forward(args)...); +#ifdef _WIN32 + child = std::make_shared(boost::process::std_out > stream_ptr_pipe->output_stream(stream::type::std_out), + boost::process::std_err > stream_ptr_pipe->output_stream(stream::type::std_err), + boost::process::std_in < stream_ptr_pipe->input_stream(), + std::forward(args)...); +#else + try { + child = std::make_shared(boost::process::std_out > stream_ptr_pipe->output_stream(stream::type::std_out), + boost::process::std_err > stream_ptr_pipe->output_stream(stream::type::std_err), + boost::process::posix::fd.bind(STDIN_FILENO, stream_ptr_pipe->child_stdin_native_handle()), + std::forward(args)...); + } + catch (...) { + stream_ptr_pipe->child_spawned(); + throw; + } + stream_ptr_pipe->child_spawned(); +#endif } return child; } diff --git a/bin/webtty/lib/rstream/webtty/server.cpp b/bin/webtty/lib/rstream/webtty/server.cpp index 69e475d..8edab97 100644 --- a/bin/webtty/lib/rstream/webtty/server.cpp +++ b/bin/webtty/lib/rstream/webtty/server.cpp @@ -602,6 +602,10 @@ class RSTREAM_GNUC_INTERNAL server::impl::session : public std::enable_shared_fr void on_send_message(const std::error_code& error_code, enum loop loop); + void do_shutdown_plain_send(); + + void on_shutdown_plain_send(const std::error_code& error_code); + void process_incoming_messages_loop(); void do_read_incoming_message(); @@ -678,6 +682,8 @@ class RSTREAM_GNUC_INTERNAL server::impl::session : public std::enable_shared_fr std::error_code m_error_code; + bool m_incoming_read_pending = false; + state_session_changed_signal_type m_state_changed_signal; std::set m_active_streams; @@ -1863,6 +1869,7 @@ void server::impl::session::do_send_error(const std::error_code& error_code) } log_session_rejected(error_code); set_state(state::disconnecting); + arm_state_timer(m_settings.m_common.m_timeouts_ms.m_close); rstream::webtty::protobuf::Message message; error::code code; if (error_code.category() == std::error_code(error::code{}).category()) { @@ -1896,6 +1903,7 @@ void server::impl::session::do_send_close(int code) } if (m_active_streams.empty()) { set_state(state::disconnecting); + arm_state_timer(m_settings.m_common.m_timeouts_ms.m_close); rstream::webtty::protobuf::Message message; message.mutable_close()->set_return_code(code); do_send_message(message, loop::exit); @@ -1957,7 +1965,7 @@ void server::impl::session::on_send_message(const std::error_code& error_code, e do_close_websocket(); } else { - on_close(error_code); + do_shutdown_plain_send(); } break; case loop::null: { @@ -1971,6 +1979,38 @@ void server::impl::session::on_send_message(const std::error_code& error_code, e } } +void server::impl::session::do_shutdown_plain_send() +{ +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif + if (m_state != state::disconnecting || m_websocket) { + return; + } +#ifdef RSTREAM_WITH_IO_STREAMS + auto completion_handler = std::bind(&session::on_shutdown_plain_send, shared_from_this(), std::placeholders::_1); + m_socket.async_shutdown_send(boost::asio::bind_executor(m_strand, completion_handler)); +#else + boost::system::error_code error_code; + m_socket.shutdown(boost::asio::socket_base::shutdown_send, error_code); + on_shutdown_plain_send(error_code); +#endif + do_read_incoming_message(); +} + +void server::impl::session::on_shutdown_plain_send(const std::error_code& error_code) +{ +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif + if (m_state != state::disconnecting) { + return; + } + if (error_code && !core::helpers::is_eof_error(error_code)) { + on_error(error_code); + } +} + void server::impl::session::process_incoming_messages_loop() { #ifdef DEBUG_BUILD @@ -1984,9 +2024,10 @@ void server::impl::session::do_read_incoming_message() #ifdef DEBUG_BUILD assert(m_strand.running_in_this_thread()); #endif - if (m_state == state::null || m_state == state::disconnected) { + if (m_state == state::null || m_state == state::disconnected || m_incoming_read_pending) { return; } + m_incoming_read_pending = true; m_buffer_socket.reset_size(); auto self = shared_from_this(); auto completion_handler = std::bind(&session::on_read_incoming_data, self, std::placeholders::_1); @@ -2008,11 +2049,18 @@ void server::impl::session::on_read_incoming_data(const std::error_code& error_c #ifdef DEBUG_BUILD assert(m_strand.running_in_this_thread()); #endif + assert(m_incoming_read_pending); + m_incoming_read_pending = false; if (m_state == state::null || m_state == state::disconnected) { return; } if (error_code) { - on_error(error_code); + if (m_state == state::disconnecting && core::helpers::is_eof_error(error_code)) { + on_close(std::error_code()); + } + else { + on_error(error_code); + } } else { rstream::webtty::protobuf::Message message; @@ -2182,6 +2230,10 @@ void server::impl::session::on_read_incoming_message(const rstream::webtty::prot error_code = error::code::unexpected_message; } else { + if (m_state == state::disconnecting) { + do_read_incoming_message(); + return; + } if (message_type == payload_type::kOpen) { protocol::config protocol_config; detail::convert(protocol_config, message.open().config()); @@ -2303,7 +2355,10 @@ void server::impl::session::on_process_data(const std::error_code& error_code) if (m_state == state::null || m_state == state::disconnected) { return; } - if (error_code) { + if (m_state == state::disconnecting) { + do_read_incoming_message(); + } + else if (error_code) { on_error(error_code); } else { diff --git a/bin/webtty/lib/rstream/webtty/stream.cpp b/bin/webtty/lib/rstream/webtty/stream.cpp index 5a59e25..988f12c 100644 --- a/bin/webtty/lib/rstream/webtty/stream.cpp +++ b/bin/webtty/lib/rstream/webtty/stream.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include @@ -40,7 +41,11 @@ ptr make_stream(const executor_type& executor, backend backend) pipe::pipe(const executor_type& executor) : base(backend::pipe), +#ifdef _WIN32 m_std_in(executor.context()), +#else + m_std_in(executor), +#endif m_std_out(executor.context()), m_std_err(executor.context()) { @@ -53,12 +58,22 @@ pipe::~pipe() void pipe::async_read_some(const boost::asio::mutable_buffer& buffer, type type, async_read_some_completion_handler&& handler) { - stream(type).async_read_some(buffer, std::move(handler)); + if (type == type::std_in) { + input_stream().async_read_some(buffer, std::move(handler)); + } + else { + output_stream(type).async_read_some(buffer, std::move(handler)); + } } void pipe::async_write(const boost::asio::const_buffer& buffer, type type, async_write_completion_handler&& handler) { - boost::asio::async_write(stream(type), buffer, std::move(handler)); + if (type == type::std_in) { + boost::asio::async_write(input_stream(), buffer, std::move(handler)); + } + else { + boost::asio::async_write(output_stream(type), buffer, std::move(handler)); + } } void pipe::close() @@ -71,22 +86,40 @@ void pipe::close() void pipe::close(type type) { boost::system::error_code tmp; - stream(type).close(tmp); + if (type == type::std_in) { + m_std_in.close(tmp); + } + else { + output_stream(type).close(tmp); + } } -pipe::stream_type& pipe::stream(type type) +pipe::stream_type& pipe::output_stream(type type) { - if (type == type::std_in) { - return m_std_in; - } - else if (type == type::std_out) { + if (type == type::std_out) { return m_std_out; } - else { + else if (type == type::std_err) { return m_std_err; } + throw std::invalid_argument("stdin is not an output stream"); } +pipe::input_stream_type& pipe::input_stream() +{ +#ifdef _WIN32 + return m_std_in; +#else + return m_std_in.stream(); +#endif +} + +#ifndef _WIN32 +int pipe::child_stdin_native_handle() { return m_std_in.child_native_handle(); } + +void pipe::child_spawned() { m_std_in.close_child_end(); } +#endif + #ifdef _WIN32 namespace { diff --git a/bin/webtty/lib/rstream/webtty/stream.hpp b/bin/webtty/lib/rstream/webtty/stream.hpp index 18e4823..25591ba 100644 --- a/bin/webtty/lib/rstream/webtty/stream.hpp +++ b/bin/webtty/lib/rstream/webtty/stream.hpp @@ -35,6 +35,9 @@ #endif #include +#ifndef _WIN32 +#include +#endif #include "error.hpp" #include "webtty.hpp" @@ -97,12 +100,25 @@ class pty { class pipe : public base { public: using stream_type = boost::process::async_pipe; +#ifdef _WIN32 + using input_stream_type = stream_type; +#else + using input_stream_type = rstream::core::posix::child_stdin::stream_type; +#endif pipe(const executor_type& executor); virtual ~pipe(); - stream_type& stream(type type); + stream_type& output_stream(type type); + + input_stream_type& input_stream(); + +#ifndef _WIN32 + int child_stdin_native_handle(); + + void child_spawned(); +#endif void async_read_some(const boost::asio::mutable_buffer& buffer, type type, async_read_some_completion_handler&& handler) override; @@ -113,7 +129,11 @@ class pipe : public base { void close(type type); private: +#ifdef _WIN32 stream_type m_std_in; +#else + rstream::core::posix::child_stdin m_std_in; +#endif stream_type m_std_out; diff --git a/lib/rstream/core/completion_handler.hpp b/lib/rstream/core/completion_handler.hpp index df6d386..74435da 100644 --- a/lib/rstream/core/completion_handler.hpp +++ b/lib/rstream/core/completion_handler.hpp @@ -152,6 +152,14 @@ void invoke_completion_handler(const Executor& ctx, Handler&& handler, Args&&... boost::asio::post(ctx, invocation_type(ctx, std::forward(handler), std::forward(args)...)); } +template +void dispatch_completion_handler(const Executor& ctx, Handler&& handler, Args&&... args) +{ + using invocation_type = detail::completion_invocation, std::decay_t, std::decay_t...>; + const auto executor = boost::asio::get_associated_executor(handler, ctx); + boost::asio::dispatch(executor, invocation_type(ctx, std::forward(handler), std::forward(args)...)); +} + } // namespace core } // namespace rstream diff --git a/lib/rstream/core/posix/child_stdin.cpp b/lib/rstream/core/posix/child_stdin.cpp new file mode 100644 index 0000000..c657a7e --- /dev/null +++ b/lib/rstream/core/posix/child_stdin.cpp @@ -0,0 +1,88 @@ +// See LICENSE file in the project root for license information. + +#include "child_stdin.hpp" + +#ifndef _WIN32 + +#include +#include + +#include +#include + +#include +#include +#include + +namespace rstream { +namespace core { +namespace posix { + +namespace { + +void set_close_on_exec(int descriptor) +{ + const auto flags = ::fcntl(descriptor, F_GETFD); + if (flags == -1 || ::fcntl(descriptor, F_SETFD, flags | FD_CLOEXEC) == -1) { + throw boost::system::system_error(errno, boost::system::system_category(), "fcntl(FD_CLOEXEC)"); + } +} + +void disable_sigpipe(int descriptor) +{ +#ifdef SO_NOSIGPIPE + constexpr int enabled = 1; + if (::setsockopt(descriptor, SOL_SOCKET, SO_NOSIGPIPE, &enabled, sizeof(enabled)) == -1) { + throw boost::system::system_error(errno, boost::system::system_category(), "setsockopt(SO_NOSIGPIPE)"); + } +#else + (void)descriptor; +#endif +} + +} // namespace + +child_stdin::child_stdin(const boost::asio::any_io_executor& executor) + : m_parent(executor), + m_child(executor) +{ + boost::asio::local::connect_pair(m_parent, m_child); + if (m_child.native_handle() == STDIN_FILENO) { + std::swap(m_parent, m_child); + } + set_close_on_exec(m_parent.native_handle()); + set_close_on_exec(m_child.native_handle()); + disable_sigpipe(m_parent.native_handle()); +} + +child_stdin::~child_stdin() +{ + boost::system::error_code ignored; + close(ignored); +} + +child_stdin::stream_type& child_stdin::stream() { return m_parent; } + +int child_stdin::child_native_handle() { return m_child.native_handle(); } + +void child_stdin::close_child_end() +{ + boost::system::error_code ignored; + m_child.close(ignored); +} + +void child_stdin::close(boost::system::error_code& error_code) +{ + error_code.clear(); + boost::system::error_code parent_error; + boost::system::error_code child_error; + m_parent.close(parent_error); + m_child.close(child_error); + error_code = parent_error ? parent_error : child_error; +} + +} // namespace posix +} // namespace core +} // namespace rstream + +#endif diff --git a/lib/rstream/core/posix/child_stdin.hpp b/lib/rstream/core/posix/child_stdin.hpp new file mode 100644 index 0000000..2c931de --- /dev/null +++ b/lib/rstream/core/posix/child_stdin.hpp @@ -0,0 +1,48 @@ +// See LICENSE file in the project root for license information. + +#pragma once + +#ifndef _WIN32 + +#include +#include + +namespace rstream { +namespace core { +namespace posix { + +// A socket pair keeps child standard-input writes on Asio's per-operation +// SIGPIPE-safe socket path without changing the process-wide signal policy. +class child_stdin { + public: + using stream_type = boost::asio::local::stream_protocol::socket; + + explicit child_stdin(const boost::asio::any_io_executor& executor); + + ~child_stdin(); + + child_stdin(const child_stdin&) = delete; + child_stdin& operator=(const child_stdin&) = delete; + + child_stdin(child_stdin&&) = delete; + child_stdin& operator=(child_stdin&&) = delete; + + stream_type& stream(); + + int child_native_handle(); + + void close_child_end(); + + void close(boost::system::error_code& error_code); + + private: + stream_type m_parent; + + stream_type m_child; +}; + +} // namespace posix +} // namespace core +} // namespace rstream + +#endif diff --git a/lib/rstream/core/windows/blocking_handle.cpp b/lib/rstream/core/windows/blocking_handle.cpp index ba4771c..24e0513 100644 --- a/lib/rstream/core/windows/blocking_handle.cpp +++ b/lib/rstream/core/windows/blocking_handle.cpp @@ -95,14 +95,12 @@ class blocking_handle::impl : public std::enable_shared_from_this { void async_read_some(const boost::asio::mutable_buffer& buffer, completion_handler&& handler) { - auto operation_allocator = boost::asio::get_associated_allocator(handler); - submit(std::allocate_shared(operation_allocator, operation::type::read, buffer, m_executor, std::move(handler))); + submit(std::make_shared(operation::type::read, buffer, m_executor, std::move(handler))); } void async_write(const boost::asio::const_buffer& buffer, completion_handler&& handler) { - auto operation_allocator = boost::asio::get_associated_allocator(handler); - submit(std::allocate_shared(operation_allocator, operation::type::write, buffer, m_executor, std::move(handler))); + submit(std::make_shared(operation::type::write, buffer, m_executor, std::move(handler))); } void cancel() diff --git a/lib/rstream/io-rstrm/acceptor.cpp b/lib/rstream/io-rstrm/acceptor.cpp index 2af38bd..f615a72 100644 --- a/lib/rstream/io-rstrm/acceptor.cpp +++ b/lib/rstream/io-rstrm/acceptor.cpp @@ -341,8 +341,7 @@ void acceptor::impl::async_accept(socket& peer, endpoint& endpoint, async_accept if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, peer, endpoint, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), peer, endpoint, std::move(handler)); { std::lock_guard lock(m_mutex); if (m_start_pending) { @@ -387,7 +386,7 @@ void acceptor::impl::async_accept(socket& peer, endpoint& endpoint, async_accept boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), op] { self->async_accept_internal(op); })); } diff --git a/lib/rstream/io-rstrm/client.cpp b/lib/rstream/io-rstrm/client.cpp index 401487a..ae42663 100644 --- a/lib/rstream/io-rstrm/client.cpp +++ b/lib/rstream/io-rstrm/client.cpp @@ -52,6 +52,10 @@ namespace rstream { namespace io_rstrm { +static constexpr unsigned int kMinControlHeartbeatIntervalMs = 1000; +static constexpr unsigned int kMaxControlHeartbeatIntervalMs = 300000; +static constexpr unsigned int kMaxControlHeartbeatTimeoutMs = 900000; + static bool invalid_published_tcp_options(const tunnel_properties& properties) { const bool published_tcp = properties.m_protocol && properties.m_protocol.value() == protocol::tcp; @@ -245,6 +249,8 @@ class RSTREAM_GNUC_INTERNAL client::impl : public std::enable_shared_from_this effective_client_details() const; void async_connect_internal(const connect_op_type::ptr& op); @@ -359,6 +365,8 @@ class RSTREAM_GNUC_INTERNAL client::impl : public std::enable_shared_from_this(operation_allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); { std::lock_guard lock(m_mutex); if (!m_is_state_non_null) { @@ -668,9 +685,8 @@ void client::impl::async_connect(async_connect_completion_handler&& handler) void client::impl::async_create_tunnel(const tunnel_properties& properties, async_create_tunnel_completion_handler&& handler) { - auto operation_allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared( - operation_allocator, + const auto op = std::allocate_shared( + core::allocator::wrapper(m_allocator), normalize_tunnel_properties(properties), std::move(handler)); auto cancellation_slot = boost::asio::get_associated_cancellation_slot(op->m_handler); @@ -697,9 +713,8 @@ void client::impl::async_create_tunnel(const tunnel_properties& properties, asyn void client::impl::async_accept_tunnel(const std::string& tunnel_id, socket& peer, endpoint& endpoint, tunnel::async_accept_completion_handler&& handler) { - auto operation_allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared( - operation_allocator, + const auto op = std::allocate_shared( + core::allocator::wrapper(m_allocator), peer, endpoint, std::move(handler)); @@ -806,6 +821,24 @@ void client::impl::arm_state_timer(unsigned int timeout_ms) task_ptr->m_timer.async_wait(completion_handler); } +void client::impl::arm_liveness_timer() +{ +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif + if (m_heartbeat_timeout_ms == 0 || m_state != state::connected) { + return; + } + const auto generation = m_generation; + m_liveness_timer.expires_after(std::chrono::milliseconds(m_heartbeat_timeout_ms)); + auto self = shared_from_this(); + m_liveness_timer.async_wait(boost::asio::bind_executor(m_strand, [self, generation](const boost::system::error_code& error_code) { + if (!error_code && generation == self->m_generation && self->m_state == state::connected) { + self->on_error(error::code::operation_timeout); + } + })); +} + void client::impl::complete_connect(const connect_op_type::ptr& op, const boost::system::error_code& error_code) { if (!op || op->m_completed) { @@ -1166,6 +1199,16 @@ void client::impl::do_open() } if (!error_code) { payload.mutable_client_details()->CopyFrom(proto_client_details); + if (m_config.m_hearbeat) { + if (m_config.m_heartbeat_interval_ms < kMinControlHeartbeatIntervalMs || m_config.m_heartbeat_interval_ms > kMaxControlHeartbeatIntervalMs) { + error_code = error::code::invalid_configuration; + } + else { + payload.mutable_liveness()->set_heartbeat_interval_ms(m_config.m_heartbeat_interval_ms); + } + } + } + if (!error_code) { message.mutable_open_control_channel_req()->CopyFrom(payload); } } @@ -1519,7 +1562,22 @@ void client::impl::on_read_incoming_message(generation_type generation, const pr if (payload.has_ok()) { const auto& ok = payload.ok(); if (!ok.client_id().empty()) { - if (ok.has_server_details()) { + m_heartbeat_timeout_ms = 0; + m_heartbeat_sequence = 0; + m_heartbeat_acknowledgement = 0; + if (ok.has_liveness()) { + const auto& liveness = ok.liveness(); + if (!m_config.m_hearbeat + || liveness.heartbeat_interval_ms() != m_config.m_heartbeat_interval_ms + || liveness.heartbeat_timeout_ms() < liveness.heartbeat_interval_ms() + || liveness.heartbeat_timeout_ms() > kMaxControlHeartbeatTimeoutMs) { + error_code = error::code::protocol_error; + } + else { + m_heartbeat_timeout_ms = liveness.heartbeat_timeout_ms(); + } + } + if (!error_code && ok.has_server_details()) { const auto& details = ok.server_details(); if (details.has_plan()) { m_status.m_plan = details.plan().value(); @@ -1534,7 +1592,9 @@ void client::impl::on_read_incoming_message(generation_type generation, const pr m_status.m_update = details.update().value(); } } - on_open(); + if (!error_code) { + on_open(); + } } else { #ifdef DEBUG_BUILD @@ -1676,6 +1736,20 @@ void client::impl::on_read_incoming_message(generation_type generation, const pr } } } + else if (message_type == payload_type::kHeartbeat) { + const auto& heartbeat = message.heartbeat(); + if (m_heartbeat_timeout_ms == 0) { + if (heartbeat.sequence() != 0 || heartbeat.acknowledgement() != 0) { + error_code = error::code::protocol_error; + } + } + else if (heartbeat.sequence() != 0 || heartbeat.acknowledgement() == 0 || heartbeat.acknowledgement() <= m_heartbeat_acknowledgement || heartbeat.acknowledgement() > m_heartbeat_sequence) { + error_code = error::code::protocol_error; + } + else { + m_heartbeat_acknowledgement = heartbeat.acknowledgement(); + } + } else if (message_type == payload_type::kCloseControlChannelRsp) { boost::system::error_code error; if (m_state != state::closing) { @@ -1691,6 +1765,9 @@ void client::impl::on_read_incoming_message(generation_type generation, const pr on_error(error_code); } else { + if (m_state == state::connected) { + arm_liveness_timer(); + } do_read_incoming_message(); } } @@ -1746,6 +1823,7 @@ void client::impl::close_internal(const boost::system::error_code& error_code) #ifdef DEBUG_BUILD m_logger->trace("closing client..."); #endif + m_liveness_timer.cancel(); set_state(state::closing); arm_state_timer(m_config.m_connection_timeout_ms); if (error_code && !m_error_code) { @@ -1791,6 +1869,7 @@ void client::impl::on_close(const boost::system::error_code& error_code) disconnection_callback = m_control_callbacks.m_on_disconnection_cb; } m_close_pending = true; + m_liveness_timer.cancel(); ++m_generation; if (m_state != state::closing) { set_state(state::closing); @@ -1828,6 +1907,9 @@ void client::impl::on_queue_cancelled( m_error_code = boost::system::error_code(); m_client_details = {}; m_status = {}; + m_heartbeat_timeout_ms = 0; + m_heartbeat_sequence = 0; + m_heartbeat_acknowledgement = 0; set_state(state::null); m_close_pending = false; { @@ -1859,7 +1941,19 @@ void client::impl::send_heartbeat() #ifdef DEBUG_BUILD assert(m_strand.running_in_this_thread()); #endif - do_send_heartbeat(); + if (m_config.m_heartbeat_interval_ms == 0 || m_state != state::connected) { + return; + } + protobuf::Message message; + auto* heartbeat = message.mutable_heartbeat(); + if (m_heartbeat_timeout_ms > 0) { + ++m_heartbeat_sequence; + if (m_heartbeat_sequence == 0) { + ++m_heartbeat_sequence; + } + heartbeat->set_sequence(m_heartbeat_sequence); + } + do_send_message(message, std::bind(&impl::do_send_heartbeat, shared_from_this())); } void client::impl::do_send_heartbeat() @@ -1905,9 +1999,7 @@ void client::impl::do_send_heartbeat() return; } if (!error_code && generation == ptr->m_generation) { - protobuf::Message message; - message.mutable_heartbeat(); - ptr->do_send_message(message, std::bind(&impl::do_send_heartbeat, ptr)); + ptr->send_heartbeat(); } task_ptr->clean(); }; diff --git a/lib/rstream/io-rstrm/definitions/rstream/io-rstrm/protobuf/messages.proto b/lib/rstream/io-rstrm/definitions/rstream/io-rstrm/protobuf/messages.proto index 7d76ede..43f24c4 100644 --- a/lib/rstream/io-rstrm/definitions/rstream/io-rstrm/protobuf/messages.proto +++ b/lib/rstream/io-rstrm/definitions/rstream/io-rstrm/protobuf/messages.proto @@ -12,7 +12,7 @@ extend google.protobuf.FieldOptions { string access = 51234; } -option (protocol_version) = "1.4.4"; +option (protocol_version) = "1.4.5"; package rstream.io_rstrm.protobuf; @@ -134,6 +134,7 @@ message TunnelProperties { message OpenControlChannelReq { ClientDetails client_details = 1; + ControlChannelLiveness liveness = 2; } // The server responds to an 'OpenControlChannelReq' message with an @@ -146,6 +147,7 @@ message OpenControlChannelRsp { message Ok { string client_id = 1; ServerDetails server_details = 2; + ControlChannelLiveness liveness = 3; } oneof payload { Ok ok = 1; @@ -261,7 +263,15 @@ message DatagramChannelClose { // Sent by client and/or server to maintain the control channel active -message Heartbeat { } +message ControlChannelLiveness { + uint32 heartbeat_interval_ms = 1; + uint32 heartbeat_timeout_ms = 2; +} + +message Heartbeat { + uint64 sequence = 1; + uint64 acknowledgement = 2; +} // Allows the server to send unsolicited messages to the client message ServerMessage { diff --git a/lib/rstream/io-rstrm/socket.cpp b/lib/rstream/io-rstrm/socket.cpp index df831ba..6da181a 100644 --- a/lib/rstream/io-rstrm/socket.cpp +++ b/lib/rstream/io-rstrm/socket.cpp @@ -58,6 +58,8 @@ class RSTREAM_GNUC_INTERNAL socket::impl : public std::enable_shared_from_thisasync_read_some(buffer, std::move(handler)); } +void socket::async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) +{ + return ptr()->async_shutdown_send(std::move(handler)); +} + void socket::adopt_impl(socket&& other) noexcept { if (this != &other) { @@ -421,8 +428,7 @@ void socket::impl::async_connect(type type, const endpoint& endpoint, async_conn if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); { std::lock_guard lock(m_mutex); if (m_is_state_non_null) { @@ -461,7 +467,7 @@ void socket::impl::async_connect(type type, const endpoint& endpoint, async_conn boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), type, endpoint, op] { self->async_connect_internal(type, endpoint, op); })); } @@ -470,13 +476,12 @@ void socket::impl::async_write_some(const boost::asio::const_buffer& buffer, asy if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); install_transfer_cancellation(op); boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), buffer, op] { self->async_write_some_internal(buffer, op); })); } @@ -485,13 +490,12 @@ void socket::impl::async_write_some(const const_buffer_sequence_type& buffer, as if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); install_transfer_cancellation(op); boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), buffer, op] { self->async_write_some_internal(buffer, op); })); } @@ -500,13 +504,12 @@ void socket::impl::async_read_some(const boost::asio::mutable_buffer& buffer, as if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); install_transfer_cancellation(op); boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), buffer, op] { self->async_read_some_internal(buffer, op); })); } @@ -515,16 +518,39 @@ void socket::impl::async_read_some(const mutable_buffer_sequence_type& buffer, a if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); install_transfer_cancellation(op); boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), buffer, op] { self->async_read_some_internal(buffer, op); })); } +void socket::impl::async_shutdown_send(async_shutdown_send_completion_handler&& handler) +{ + if (!handler) { + return; + } + boost::asio::dispatch( + m_strand, + core::bind_handler_allocator( + core::allocator::wrapper(m_allocator), + [self = shared_from_this(), handler = std::move(handler)]() mutable { + if (self->m_state != state::connected) { + rstream::core::invoke_completion_handler(self->m_executor, std::move(handler), error::make_error_code(error::code::invalid_state)); + return; + } +#ifdef RSTREAM_WITH_IO_STREAMS + self->m_socket.async_shutdown_send(rstream::core::bind_handler_lifetime(self, std::move(handler))); +#else + boost::system::error_code error_code; + self->m_socket.shutdown(boost::asio::socket_base::shutdown_send, error_code); + rstream::core::invoke_completion_handler(self->m_executor, std::move(handler), error_code); +#endif + })); +} + void socket::impl::install_transfer_cancellation(const transfer_op::ptr& op) { auto cancellation_slot = boost::asio::get_associated_cancellation_slot(op->m_handler); @@ -826,9 +852,7 @@ void socket::impl::do_connect(const resolver_type::results_type& results) }; auto internal_handler = boost::asio::bind_executor( m_strand, - core::bind_handler_allocator( - core::allocator::wrapper(m_allocator), - boost::asio::bind_cancellation_slot(boost::asio::cancellation_slot(), std::move(completion_handler)))); + boost::asio::bind_allocator(core::allocator::wrapper(m_allocator), std::move(completion_handler))); boost::asio::async_connect(m_socket, results, std::move(internal_handler)); } diff --git a/lib/rstream/io-rstrm/socket.hpp b/lib/rstream/io-rstrm/socket.hpp index c2c3847..78cd660 100644 --- a/lib/rstream/io-rstrm/socket.hpp +++ b/lib/rstream/io-rstrm/socket.hpp @@ -74,6 +74,8 @@ class socket : public io::stream_socket_base { void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler) override; + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) override; + void adopt_impl(socket&& other) noexcept; std::shared_ptr ptr(); diff --git a/lib/rstream/io/acceptor_base.hpp b/lib/rstream/io/acceptor_base.hpp index 2f5c79a..63e9fc6 100644 --- a/lib/rstream/io/acceptor_base.hpp +++ b/lib/rstream/io/acceptor_base.hpp @@ -80,8 +80,7 @@ class acceptor_base : public socket_base { [this](auto&& handler) { using operation_type = owning_accept_operation>; auto executor = io_object::get_executor(); - auto allocator = boost::asio::get_associated_allocator(handler); - auto operation = std::allocate_shared(allocator, executor, std::forward(handler)); + auto operation = std::make_shared(executor, std::forward(handler)); auto cancellation_slot = boost::asio::get_associated_cancellation_slot(operation->m_handler); if (cancellation_slot.is_connected()) { const std::weak_ptr weak_operation = operation; diff --git a/lib/rstream/io/detail/stream/acceptor_impl.hpp b/lib/rstream/io/detail/stream/acceptor_impl.hpp index 1c2d750..83f433f 100644 --- a/lib/rstream/io/detail/stream/acceptor_impl.hpp +++ b/lib/rstream/io/detail/stream/acceptor_impl.hpp @@ -211,9 +211,7 @@ native_endpoint_type acceptor_impl void acceptor_impl::async_accept_internal(stream_socket& peer, endpoint& endpoint, async_accept_completion_handler&& handler) { - const auto allocator = boost::asio::get_associated_allocator(handler); - std::allocate_shared( - allocator, + std::make_shared( std::enable_shared_from_this::shared_from_this(), peer, endpoint, m_url, std::move(handler)) ->run(); diff --git a/lib/rstream/io/detail/stream/acceptor_ssl.cpp b/lib/rstream/io/detail/stream/acceptor_ssl.cpp index 7f316ff..9e0bbd5 100644 --- a/lib/rstream/io/detail/stream/acceptor_ssl.cpp +++ b/lib/rstream/io/detail/stream/acceptor_ssl.cpp @@ -211,12 +211,11 @@ endpoint acceptor_ssl::impl::local_endpoint(boost::system::error_code& error_cod void acceptor_ssl::impl::async_accept(stream_socket& peer, endpoint& endpoint, async_accept_completion_handler&& handler) { - const auto allocator = boost::asio::get_associated_allocator(handler); - auto self = shared_from_this(); + auto self = shared_from_this(); boost::asio::dispatch( m_strand, boost::asio::bind_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = std::move(self), peer = &peer, endpoint = &endpoint, handler = std::move(handler)]() mutable { self->async_accept_internal(*peer, *endpoint, std::move(handler)); })); @@ -261,8 +260,8 @@ void acceptor_ssl::impl::async_accept_internal(stream_socket& peer, endpoint& en it = m_async_accept_upstream_ops.erase(it); } else { - const auto allocator = boost::asio::get_associated_allocator(handler); - m_async_accept_downstream_op = std::allocate_shared(allocator, peer, endpoint, std::move(handler)); + m_async_accept_downstream_op = std::allocate_shared( + core::allocator::wrapper(m_allocator), peer, endpoint, std::move(handler)); do_accept(); } } diff --git a/lib/rstream/io/detail/stream/stream_socket.cpp b/lib/rstream/io/detail/stream/stream_socket.cpp index 37e311c..c5438bf 100644 --- a/lib/rstream/io/detail/stream/stream_socket.cpp +++ b/lib/rstream/io/detail/stream/stream_socket.cpp @@ -44,6 +44,8 @@ class RSTREAM_GNUC_INTERNAL stream_socket::impl { void async_read_some(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler); + void async_shutdown_send(async_shutdown_send_completion_handler&& handler); + stream_socket_const_ptr native_handle() const; stream_socket_ptr native_handle(); @@ -149,6 +151,11 @@ void stream_socket::async_read_some_internal(const mutable_buffer_sequence_type& m_impl->async_read_some(buffer, std::move(handler)); } +void stream_socket::async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) +{ + m_impl->async_shutdown_send(std::move(handler)); +} + stream_socket::impl::impl(const executor_type& executor) : m_executor(executor) { @@ -270,6 +277,15 @@ void stream_socket::impl::async_read_some(const mutable_buffer_sequence_type& bu } } +void stream_socket::impl::async_shutdown_send(async_shutdown_send_completion_handler&& handler) +{ + if (!initialized()) { + rstream::core::invoke_completion_handler(m_executor, std::move(handler), error::make_error_code(error::code::uninitialized_object)); + return; + } + m_native_handle->async_shutdown_send(std::move(handler)); +} + stream_socket_const_ptr stream_socket::impl::native_handle() const { return m_native_handle; diff --git a/lib/rstream/io/detail/stream/stream_socket.hpp b/lib/rstream/io/detail/stream/stream_socket.hpp index 4caa320..8a74329 100644 --- a/lib/rstream/io/detail/stream/stream_socket.hpp +++ b/lib/rstream/io/detail/stream/stream_socket.hpp @@ -69,6 +69,8 @@ class stream_socket : public stream_socket_base { void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler) override; + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) override; + std::shared_ptr m_impl; }; diff --git a/lib/rstream/io/detail/stream/stream_socket_impl.hpp b/lib/rstream/io/detail/stream/stream_socket_impl.hpp index d76bf53..d503588 100644 --- a/lib/rstream/io/detail/stream/stream_socket_impl.hpp +++ b/lib/rstream/io/detail/stream/stream_socket_impl.hpp @@ -3,6 +3,7 @@ #pragma once #include +#include #include #include @@ -68,6 +69,8 @@ class stream_socket_impl : public stream_socket_base, public object_ba void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler) override; + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) override; + native_socket_type m_socket; const endpoint_base::protocol_type::value_type m_protocol; @@ -241,6 +244,29 @@ void stream_socket_impl::async_read_so std::move(handler))); } +template +void stream_socket_impl::async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) +{ + if constexpr (requires(native_socket_type& socket, async_shutdown_send_completion_handler&& completion_handler) { + socket.async_shutdown_send(std::move(completion_handler)); + }) { + m_socket.async_shutdown_send( + rstream::core::bind_handler_lifetime( + std::enable_shared_from_this::shared_from_this(), + std::move(handler))); + } + else if constexpr (requires(native_socket_type& socket, boost::system::error_code& error_code) { + socket.shutdown(boost::asio::socket_base::shutdown_send, error_code); + }) { + boost::system::error_code error_code; + m_socket.shutdown(boost::asio::socket_base::shutdown_send, error_code); + rstream::core::invoke_completion_handler(get_executor(), std::move(handler), error_code); + } + else { + rstream::core::invoke_completion_handler(get_executor(), std::move(handler), boost::asio::error::operation_not_supported); + } +} + } // namespace stream } // namespace detail } // namespace io diff --git a/lib/rstream/io/detail/stream/stream_socket_ssl.cpp b/lib/rstream/io/detail/stream/stream_socket_ssl.cpp index 72e234b..b4c9e54 100644 --- a/lib/rstream/io/detail/stream/stream_socket_ssl.cpp +++ b/lib/rstream/io/detail/stream/stream_socket_ssl.cpp @@ -123,6 +123,8 @@ class RSTREAM_GNUC_INTERNAL stream_socket_ssl::impl : public std::enable_shared_ void async_shutdown(async_shutdown_completion_handler&& handler); + void async_shutdown_send(async_shutdown_send_completion_handler&& handler); + void async_connect(const endpoint& endpoint, async_connect_completion_handler&& handler); void async_write_some(const boost::asio::const_buffer& buffer, async_write_some_completion_handler&& handler); @@ -143,6 +145,8 @@ class RSTREAM_GNUC_INTERNAL stream_socket_ssl::impl : public std::enable_shared_ void async_shutdown_internal(async_shutdown_completion_handler&& handler); + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler); + void async_connect_internal(const endpoint& endpoint, async_connect_completion_handler&& handler); void async_write_some_internal(const boost::asio::const_buffer& buffer, async_write_some_completion_handler&& handler); @@ -154,6 +158,24 @@ class RSTREAM_GNUC_INTERNAL stream_socket_ssl::impl : public std::enable_shared_ void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler); #endif + template + auto bind_ssl_handler(Handler&& handler) + { + auto completion_handler = rstream::core::bind_associated_handler( + std::forward(handler), + [executor = m_next_layer->get_executor()](auto& associated_handler, auto&&... args) mutable { + rstream::core::dispatch_completion_handler( + executor, + std::move(associated_handler), + std::forward(args)...); + }); +#if SSL_STREAM_USE_STRAND == 1 + return boost::asio::bind_executor(m_strand, std::move(completion_handler)); +#else + return completion_handler; +#endif + } + boost::asio::ssl::context make_ssl_context(); #if SSL_STREAM_USE_OPENSSL_ENGINE == 1 @@ -341,6 +363,11 @@ void stream_socket_ssl::async_read_some_internal(const mutable_buffer_sequence_t m_impl->async_read_some(buffer, std::move(handler)); } +void stream_socket_ssl::async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) +{ + m_impl->async_shutdown_send(std::move(handler)); +} + stream_socket_ssl::impl::impl(stream_socket_ptr next_layer, const ssl::config& config, type type, core::allocator::ptr allocator) : #if SSL_STREAM_USE_STRAND == 1 @@ -413,6 +440,13 @@ void stream_socket_ssl::impl::async_shutdown(async_shutdown_completion_handler&& } #endif +#if SSL_STREAM_USE_STRAND == 1 +void stream_socket_ssl::impl::async_shutdown_send(async_shutdown_send_completion_handler&& handler) +{ + boost::asio::dispatch(m_strand, std::bind_front(&impl::async_shutdown_send_internal, shared_from_this(), std::move(handler))); +} +#endif + #if SSL_STREAM_USE_STRAND == 1 void stream_socket_ssl::impl::async_connect(const endpoint& endpoint, async_connect_completion_handler&& handler) { @@ -480,7 +514,7 @@ void stream_socket_ssl::impl:: #endif m_ssl_stream.async_handshake( handshake_type, - rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler))); + bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); } void stream_socket_ssl::impl:: @@ -496,8 +530,59 @@ void stream_socket_ssl::impl:: assert(m_strand.running_in_this_thread()); #endif #endif - auto operation_allocator = boost::asio::get_associated_allocator(handler); - std::allocate_shared(operation_allocator, shared_from_this(), std::forward(handler))->run(); + std::allocate_shared( + core::allocator::wrapper(m_allocator), + shared_from_this(), + std::forward(handler)) + ->run(); +} + +void stream_socket_ssl::impl:: +#if SSL_STREAM_USE_STRAND == 1 + async_shutdown_send_internal +#else + async_shutdown_send +#endif + (async_shutdown_send_completion_handler&& handler) +{ +#if SSL_STREAM_USE_STRAND == 1 +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif +#endif + auto* ssl = m_ssl_stream.native_handle(); + const auto previous_state = ::SSL_get_shutdown(ssl); + const auto previous_receive = previous_state & SSL_RECEIVED_SHUTDOWN; + if ((previous_state & SSL_SENT_SHUTDOWN) != 0) { + rstream::core::invoke_completion_handler(m_next_layer->get_executor(), std::move(handler), boost::system::error_code()); + return; + } + ::SSL_set_shutdown(ssl, previous_state | SSL_RECEIVED_SHUTDOWN); + auto completion_handler = rstream::core::bind_associated_handler( + std::move(handler), + [self = shared_from_this(), previous_receive, executor = m_next_layer->get_executor()](auto& associated_handler, const boost::system::error_code& error_code) mutable { +#if SSL_STREAM_USE_STRAND == 1 +#ifdef DEBUG_BUILD + assert(self->m_strand.running_in_this_thread()); +#endif +#endif + auto* native_ssl = self->m_ssl_stream.native_handle(); + const auto shutdown_state = ::SSL_get_shutdown(native_ssl); + ::SSL_set_shutdown(native_ssl, (shutdown_state & ~SSL_RECEIVED_SHUTDOWN) | previous_receive); + rstream::core::dispatch_completion_handler(executor, std::move(associated_handler), error_code); + }); + auto lifetime_handler = rstream::core::bind_handler_lifetime(shared_from_this(), std::move(completion_handler)); + try { +#if SSL_STREAM_USE_STRAND == 1 + m_ssl_stream.async_shutdown(boost::asio::bind_executor(m_strand, std::move(lifetime_handler))); +#else + m_ssl_stream.async_shutdown(std::move(lifetime_handler)); +#endif + } + catch (...) { + ::SSL_set_shutdown(ssl, previous_state); + throw; + } } void stream_socket_ssl::impl:: @@ -513,8 +598,12 @@ void stream_socket_ssl::impl:: assert(m_strand.running_in_this_thread()); #endif #endif - auto operation_allocator = boost::asio::get_associated_allocator(handler); - std::allocate_shared(operation_allocator, shared_from_this(), endpoint, std::forward(handler))->run(); + std::allocate_shared( + core::allocator::wrapper(m_allocator), + shared_from_this(), + endpoint, + std::forward(handler)) + ->run(); } void stream_socket_ssl::impl:: @@ -532,7 +621,7 @@ void stream_socket_ssl::impl:: #endif m_ssl_stream.async_write_some( buffer, - rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler))); + bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); } void stream_socket_ssl::impl:: @@ -550,7 +639,7 @@ void stream_socket_ssl::impl:: #endif m_ssl_stream.async_write_some( buffer, - rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler))); + bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); } void stream_socket_ssl::impl:: @@ -568,7 +657,7 @@ void stream_socket_ssl::impl:: #endif m_ssl_stream.async_read_some( buffer, - rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler))); + bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); } void stream_socket_ssl::impl:: @@ -586,7 +675,7 @@ void stream_socket_ssl::impl:: #endif m_ssl_stream.async_read_some( buffer, - rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler))); + bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); } boost::asio::ssl::context stream_socket_ssl::impl::make_ssl_context() @@ -1733,12 +1822,22 @@ void stream_socket_ssl::impl::async_connect_operation::do_connect() on_complete(error_code); } else { - m_ptr->m_next_layer->async_connect(m_endpoint, std::bind(&async_connect_operation::on_connect, shared_from_this(), std::placeholders::_1)); + auto completion_handler = std::bind(&async_connect_operation::on_connect, shared_from_this(), std::placeholders::_1); +#if SSL_STREAM_USE_STRAND == 1 + m_ptr->m_next_layer->async_connect(m_endpoint, boost::asio::bind_executor(m_ptr->m_strand, std::move(completion_handler))); +#else + m_ptr->m_next_layer->async_connect(m_endpoint, std::move(completion_handler)); +#endif } } void stream_socket_ssl::impl::async_connect_operation::on_connect(const boost::system::error_code& error_code) { +#if SSL_STREAM_USE_STRAND == 1 +#ifdef DEBUG_BUILD + assert(m_ptr->m_strand.running_in_this_thread()); +#endif +#endif if (error_code) { on_complete(error_code); } @@ -1752,11 +1851,21 @@ void stream_socket_ssl::impl::async_connect_operation::do_handshake() #ifdef DEBUG_BUILD m_logger->trace("handshaking with SSL server..."); #endif - m_ptr->async_handshake(std::bind(&async_connect_operation::on_handshake, shared_from_this(), std::placeholders::_1)); + auto completion_handler = std::bind(&async_connect_operation::on_handshake, shared_from_this(), std::placeholders::_1); +#if SSL_STREAM_USE_STRAND == 1 + m_ptr->async_handshake(boost::asio::bind_executor(m_ptr->m_strand, std::move(completion_handler))); +#else + m_ptr->async_handshake(std::move(completion_handler)); +#endif } void stream_socket_ssl::impl::async_connect_operation::on_handshake(const boost::system::error_code& error_code) { +#if SSL_STREAM_USE_STRAND == 1 +#ifdef DEBUG_BUILD + assert(m_ptr->m_strand.running_in_this_thread()); +#endif +#endif #ifdef DEBUG_BUILD m_logger->trace("handshake with SSL server completed [error_code: {}]", (error_code ? error_code.message() : "none")); #endif diff --git a/lib/rstream/io/detail/stream/stream_socket_ssl.hpp b/lib/rstream/io/detail/stream/stream_socket_ssl.hpp index e67bf48..fde66f1 100644 --- a/lib/rstream/io/detail/stream/stream_socket_ssl.hpp +++ b/lib/rstream/io/detail/stream/stream_socket_ssl.hpp @@ -63,6 +63,8 @@ class stream_socket_ssl : public stream_socket_base { void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler) override; + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) override; + std::shared_ptr m_impl; }; diff --git a/lib/rstream/io/queue.hpp b/lib/rstream/io/queue.hpp index c3c86b4..ab8840f 100644 --- a/lib/rstream/io/queue.hpp +++ b/lib/rstream/io/queue.hpp @@ -301,11 +301,10 @@ typename queue::executor_type queue::impl::get_executor() template void queue::impl::async_send(const core::buffer buffer, async_send_completion_handler&& handler) { - auto allocator = boost::asio::get_associated_allocator(handler); - auto self = std::enable_shared_from_this::shared_from_this(); - auto task_ptr = std::allocate_shared(allocator, buffer, std::move(handler)); + auto self = std::enable_shared_from_this::shared_from_this(); + auto task_ptr = std::allocate_shared(core::allocator::wrapper(m_allocator), buffer, std::move(handler)); task_ptr->arm(task_ptr, self, m_strand); - boost::asio::dispatch(m_strand, boost::asio::bind_allocator(allocator, [self, task_ptr] { self->send(task_ptr); })); + boost::asio::dispatch(m_strand, boost::asio::bind_allocator(core::allocator::wrapper(m_allocator), [self, task_ptr] { self->send(task_ptr); })); } template @@ -318,12 +317,11 @@ void queue::impl::cancel() template void queue::impl::async_cancel(async_cancel_completion_handler&& handler) { - auto allocator = boost::asio::get_associated_allocator(handler); - auto self = std::enable_shared_from_this::shared_from_this(); + auto self = std::enable_shared_from_this::shared_from_this(); boost::asio::dispatch( m_strand, boost::asio::bind_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self, handler = std::move(handler)]() mutable { self->cancel_internal(std::move(handler)); })); diff --git a/lib/rstream/io/stream_socket_base.hpp b/lib/rstream/io/stream_socket_base.hpp index d828344..55b3efc 100644 --- a/lib/rstream/io/stream_socket_base.hpp +++ b/lib/rstream/io/stream_socket_base.hpp @@ -48,7 +48,29 @@ class stream_socket_base_interface { buffers); } + using async_shutdown_send_completion_handler = rstream::core::completion_handler; + template + auto async_shutdown_send(BOOST_ASIO_MOVE_ARG(shutdown_handler) handler) + { + return boost::asio::async_initiate( + async_shutdown_send_op(*this), handler); + } + private: + struct async_shutdown_send_op { + async_shutdown_send_op(stream_socket_base_interface& socket) + : m_socket(socket) + { + } + template + void operator()(BOOST_ASIO_MOVE_ARG(shutdown_handler) handler) + { + auto executor = boost::asio::get_associated_executor(handler, m_socket.m_executor); + m_socket.async_shutdown_send_internal(boost::asio::bind_executor(executor, std::forward(handler))); + } + stream_socket_base_interface& m_socket; + }; + template struct async_write_some_op { async_write_some_op(stream_socket_base_interface& socket) @@ -129,6 +151,8 @@ class stream_socket_base_interface { virtual void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler) = 0; + virtual void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) = 0; + io_object::executor_type m_executor; }; diff --git a/test/core/common/test_core_executor_binder.cpp b/test/core/common/test_core_executor_binder.cpp index 4363181..2fbbe00 100644 --- a/test/core/common/test_core_executor_binder.cpp +++ b/test/core/common/test_core_executor_binder.cpp @@ -180,6 +180,45 @@ static void check_completion_is_deferred_on_associated_executor() assert(allocation->m_allocations == allocation->m_deallocations); } +static void check_completion_can_dispatch_on_associated_executor() +{ + boost::asio::io_context fallback_context; + boost::asio::io_context associated_context; + auto strand = boost::asio::make_strand(associated_context); + auto allocation = std::make_shared(); + boost::asio::cancellation_signal cancellation; + std::atomic_size_t calls = 0; + auto work = boost::asio::make_work_guard(associated_context); + boost::asio::post(strand, [&] { + rstream::core::completion_handler handler( + associated_handler(strand, counting_allocator(allocation), cancellation.slot(), calls)); + rstream::core::dispatch_completion_handler(fallback_context.get_executor(), std::move(handler), boost::system::error_code()); + assert(calls == 1); + work.reset(); + }); + associated_context.run(); + assert(calls == 1); + assert(allocation->m_allocations == allocation->m_deallocations); +} + +static void check_abandoned_completion_releases_associated_state() +{ + auto allocation = std::make_shared(); + std::atomic_size_t calls = 0; + { + boost::asio::io_context associated_context; + boost::asio::io_context fallback_context; + auto strand = boost::asio::make_strand(associated_context); + boost::asio::cancellation_signal cancellation; + rstream::core::completion_handler handler( + associated_handler(strand, counting_allocator(allocation), cancellation.slot(), calls)); + rstream::core::invoke_completion_handler(fallback_context.get_executor(), std::move(handler), boost::system::error_code()); + } + assert(calls == 0); + assert(allocation->m_allocations > 0); + assert(allocation->m_allocations == allocation->m_deallocations); +} + static void check_adapter_preserves_associations() { boost::asio::io_context fallback_context; @@ -320,6 +359,8 @@ int main() { check_type_erasure_preserves_associations(); check_completion_is_deferred_on_associated_executor(); + check_completion_can_dispatch_on_associated_executor(); + check_abandoned_completion_releases_associated_state(); check_adapter_preserves_associations(); check_lifetime_adapter_preserves_owner_and_associations(); check_strand_serializes_type_erased_handlers(); diff --git a/test/io/common/test_io_common_queue.cpp b/test/io/common/test_io_common_queue.cpp index 824e730..81f50d4 100644 --- a/test/io/common/test_io_common_queue.cpp +++ b/test/io/common/test_io_common_queue.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -26,6 +27,54 @@ #include #include +struct allocator_state { + std::atomic_size_t m_allocations = 0; + std::atomic_size_t m_deallocations = 0; +}; + +template +class stateful_allocator { + public: + using value_type = T; + + explicit stateful_allocator(std::shared_ptr state) + : m_state(std::move(state)) + { + } + + template + stateful_allocator(const stateful_allocator& other) + : m_state(other.state()) + { + } + + T* allocate(std::size_t count) + { + ++m_state->m_allocations; + return std::allocator().allocate(count); + } + + void deallocate(T* pointer, std::size_t count) + { + ++m_state->m_deallocations; + std::allocator().deallocate(pointer, count); + } + + const std::shared_ptr& state() const + { + return m_state; + } + + template + bool operator==(const stateful_allocator& other) const + { + return m_state == other.state(); + } + + private: + std::shared_ptr m_state; +}; + class controlled_transport { public: using executor_type = boost::asio::io_context::executor_type; @@ -446,6 +495,23 @@ static void check_deferred_operations_are_lazy() assert(cancel_calls == 1); } +static void check_abandoned_type_erased_operations_are_safe() +{ + auto allocation = std::make_shared(); + { + boost::asio::io_context io_context; + controlled_transport transport(io_context.get_executor()); + rstream::io::queue queue(transport); + queue.async_send( + make_buffer(1), + boost::asio::bind_allocator(stateful_allocator(allocation), [](const boost::system::error_code&) {})); + queue.async_cancel( + boost::asio::bind_allocator(stateful_allocator(allocation), [](const boost::system::error_code&) {})); + } + assert(allocation->m_allocations > 0); + assert(allocation->m_allocations == allocation->m_deallocations); +} + int main() { check_owned_move_only_transport(); @@ -457,5 +523,6 @@ int main() check_async_cancel_waits_for_active_send(); check_async_cancel_supports_multiple_waiters(); check_deferred_operations_are_lazy(); + check_abandoned_type_erased_operations_are_safe(); return 0; } diff --git a/test/io/common/test_io_common_stream_tls.cpp b/test/io/common/test_io_common_stream_tls.cpp index 8ed7d21..35e4c7d 100644 --- a/test/io/common/test_io_common_stream_tls.cpp +++ b/test/io/common/test_io_common_stream_tls.cpp @@ -5,10 +5,12 @@ #include #include #include +#include #include #include #include #include +#include #include #include #include @@ -16,6 +18,7 @@ #include #include +#include #include #include #include @@ -200,6 +203,11 @@ class fake_stream_socket : public rstream::io::detail::stream::stream_socket_int rstream::core::invoke_completion_handler(get_executor(), std::move(handler), boost::asio::error::would_block, boost::asio::buffer_size(buffer)); } + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) override + { + rstream::core::invoke_completion_handler(get_executor(), std::move(handler), boost::system::error_code()); + } + endpoint_type m_endpoint; }; @@ -636,6 +644,224 @@ static void check_tls_accept_connect_and_transfer(const std::string& groups_quer assert(server_sequence_sent); } +class tls_half_close_exchange : public std::enable_shared_from_this { + public: + tls_half_close_exchange(rstream::io::stream::stream_socket& client, rstream::io::stream::stream_socket& server) + : m_client(client), + m_server(server), + m_payload(1024 * 1024, '\0'), + m_response("response-after-client-eof"), + m_client_response(m_response.size(), '\0') + { + for (std::size_t i = 0; i < m_payload.size(); ++i) { + m_payload[i] = static_cast((i * 31U + 17U) % 251U); + } + } + + void run() + { + read_server(); + boost::asio::async_read( + m_client, + boost::asio::buffer(m_client_response), + [self = shared_from_this()](const boost::system::error_code& error_code, std::size_t size) { + self->m_client_read_error = error_code; + self->m_client_read_size = size; + self->m_client_read_done = true; + self->mark_completed(g_client_read_completed); + }); + boost::asio::async_write( + m_client, + boost::asio::buffer(m_payload), + [self = shared_from_this()](const boost::system::error_code& error_code, std::size_t size) { + self->m_client_write_error = error_code; + self->m_client_write_size = size; + self->m_client_write_done = true; + self->mark_completed(g_client_write_completed); + self->m_client.async_shutdown_send( + [self](const boost::system::error_code& shutdown_error) { + self->on_shutdown_send(shutdown_error); + }); + }); + } + + bool completed() const + { + return m_completion_mask.load(std::memory_order_acquire) == g_all_completed; + } + + bool wait_for_completion(std::chrono::milliseconds timeout) + { + std::unique_lock lock(m_completion_mutex); + return m_completion_condition.wait_for(lock, timeout, [this] { return completed(); }); + } + + void assert_success() const + { + assert(!m_client_write_error); + assert(m_client_write_size == m_payload.size()); + assert(m_server_read_error == boost::asio::error::eof); + assert(m_server_payload == m_payload); + assert(!m_server_write_error); + assert(m_server_write_size == m_response.size()); + assert(!m_client_read_error); + assert(m_client_read_size == m_response.size()); + assert(m_client_response == m_response); + assert(!m_first_shutdown_error); + assert(!m_second_shutdown_error); + assert(m_shutdown_count == 2); + } + + private: + void read_server() + { + m_server.async_read_some( + boost::asio::buffer(m_server_buffer), + [self = shared_from_this()](const boost::system::error_code& error_code, std::size_t size) { + self->m_server_payload.append(self->m_server_buffer.data(), size); + if (!error_code) { + self->read_server(); + return; + } + self->m_server_read_error = error_code; + self->m_server_read_done = true; + self->mark_completed(g_server_read_completed); + if (error_code == boost::asio::error::eof) { + boost::asio::async_write( + self->m_server, + boost::asio::buffer(self->m_response), + [self](const boost::system::error_code& write_error, std::size_t write_size) { + self->m_server_write_error = write_error; + self->m_server_write_size = write_size; + self->m_server_write_done = true; + self->mark_completed(g_server_write_completed); + }); + } + else { + self->m_server_write_done = true; + self->mark_completed(g_server_write_completed); + } + }); + } + + void on_shutdown_send(const boost::system::error_code& error_code) + { + ++m_shutdown_count; + if (m_shutdown_count == 1) { + m_first_shutdown_error = error_code; + m_client.async_shutdown_send( + [self = shared_from_this()](const boost::system::error_code& repeated_error) { + self->on_shutdown_send(repeated_error); + }); + } + else { + m_second_shutdown_error = error_code; + mark_completed(g_shutdown_completed); + } + } + + void mark_completed(unsigned int bit) + { + const auto previous = m_completion_mask.fetch_or(bit, std::memory_order_acq_rel); + assert((previous & bit) == 0); + if ((previous | bit) == g_all_completed) { + std::lock_guard lock(m_completion_mutex); + m_completion_condition.notify_all(); + } + } + + static constexpr unsigned int g_client_write_completed = 1U << 0U; + static constexpr unsigned int g_shutdown_completed = 1U << 1U; + static constexpr unsigned int g_server_read_completed = 1U << 2U; + static constexpr unsigned int g_server_write_completed = 1U << 3U; + static constexpr unsigned int g_client_read_completed = 1U << 4U; + static constexpr unsigned int g_all_completed = g_client_write_completed | g_shutdown_completed | g_server_read_completed | g_server_write_completed | g_client_read_completed; + + rstream::io::stream::stream_socket& m_client; + rstream::io::stream::stream_socket& m_server; + std::string m_payload; + std::string m_response; + std::string m_client_response; + std::string m_server_payload; + std::array m_server_buffer{}; + boost::system::error_code m_client_write_error; + boost::system::error_code m_server_read_error; + boost::system::error_code m_server_write_error; + boost::system::error_code m_client_read_error; + boost::system::error_code m_first_shutdown_error; + boost::system::error_code m_second_shutdown_error; + std::size_t m_client_write_size = 0; + std::size_t m_server_write_size = 0; + std::size_t m_client_read_size = 0; + unsigned int m_shutdown_count = 0; + std::atomic_uint m_completion_mask = 0; + std::mutex m_completion_mutex; + std::condition_variable m_completion_condition; + bool m_client_write_done = false; + bool m_server_read_done = false; + bool m_server_write_done = false; + bool m_client_read_done = false; +}; + +static void check_tls_half_close_preserves_receive_direction() +{ + certificate_files files; + boost::asio::io_context io_context; + const auto port = unused_tcp_port(); + const auto server_endpoint = resolve_one( + io_context, + "tcp://127.0.0.1:" + std::to_string(port) + "?ssl&ssl.cert_file=" + files.cert_file() + + "&ssl.key_file=" + files.key_file() + + "&ssl.peer_verification=false&ssl.request_peer_cert=false&ssl.async_shutdown_timeout_ms=0"); + const auto client_endpoint = resolve_one( + io_context, + "tcp://127.0.0.1:" + std::to_string(port) + + "?ssl&ssl.peer_verification=false&ssl.request_peer_cert=false&ssl.sni=localhost&ssl.async_shutdown_timeout_ms=0"); + rstream::io::stream::acceptor acceptor(io_context.get_executor()); + boost::system::error_code error_code; + acceptor.open(server_endpoint, error_code); + assert(!error_code); + acceptor.bind(server_endpoint, error_code); + assert(!error_code); + acceptor.listen(boost::asio::socket_base::max_listen_connections, error_code); + assert(!error_code); + rstream::io::stream::stream_socket server(io_context.get_executor()); + rstream::io::stream::stream_socket client(io_context.get_executor()); + rstream::io::stream::endpoint remote_endpoint; + bool accepted = false; + bool connected = false; + acceptor.async_accept(server, remote_endpoint, [&](const boost::system::error_code& error) { + assert(!error); + accepted = true; + }); + client.async_connect(client_endpoint, [&](const boost::system::error_code& error) { + assert(!error); + connected = true; + }); + run_until(io_context, [&] { return accepted && connected; }); + auto exchange = std::make_shared(client, server); + exchange->run(); + auto work = boost::asio::make_work_guard(io_context); + std::array workers; + for (auto& worker : workers) { + worker = std::thread([&] { io_context.run(); }); + } + const auto completed = exchange->wait_for_completion(std::chrono::seconds(10)); + boost::system::error_code ignored; + client.close(ignored); + server.close(ignored); + acceptor.close(ignored); + work.reset(); + if (!completed) { + io_context.stop(); + } + for (auto& worker : workers) { + worker.join(); + } + assert(completed); + exchange->assert_success(); +} + static void check_tls_shutdown_timeout_is_serialized() { certificate_files files; @@ -825,6 +1051,7 @@ int main(int argc, char** argv) check_tls_config_errors(); check_direct_tls_context_configuration(); check_tls_accept_connect_and_transfer(); + check_tls_half_close_preserves_receive_direction(); check_tls_shutdown_timeout_is_serialized(); check_tls_accept_preserves_peer_executor(); check_tls_peer_verification_checks_hostname(); diff --git a/test/io/io-rstrm/test_io_rstrm_control_channel.cpp b/test/io/io-rstrm/test_io_rstrm_control_channel.cpp index 4e79635..3253bbd 100644 --- a/test/io/io-rstrm/test_io_rstrm_control_channel.cpp +++ b/test/io/io-rstrm/test_io_rstrm_control_channel.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include @@ -169,6 +170,19 @@ static void wait_for_peer_close(tcp::socket& socket) } } +static protobuf::Message read_control_message_and_acknowledge_heartbeats(tcp::socket& socket) +{ + for (;;) { + auto message = read_message(socket); + if (!message.has_heartbeat()) { + return message; + } + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(message.heartbeat().sequence()); + write_message(socket, acknowledgement); + } +} + static void wait_until_ready(const std::atomic_bool& ready, const std::string& timeout_message) { const auto deadline = std::chrono::steady_clock::now() + kFakeEngineIoTimeout; @@ -191,6 +205,15 @@ static protobuf::Message open_control_response() return response; } +static protobuf::Message open_control_liveness_response(std::uint32_t heartbeat_interval_ms, std::uint32_t heartbeat_timeout_ms) +{ + auto response = open_control_response(); + auto* liveness = response.mutable_open_control_channel_rsp()->mutable_ok()->mutable_liveness(); + liveness->set_heartbeat_interval_ms(heartbeat_interval_ms); + liveness->set_heartbeat_timeout_ms(heartbeat_timeout_ms); + return response; +} + static protobuf::Message open_tunnel_response(const protobuf::OpenTunnelReq& request) { protobuf::Message response; @@ -214,6 +237,10 @@ static protobuf::Message close_control_response() return response; } +static protobuf::Message proxy_connection_request(const std::string& stream_id, const std::string& tunnel_id); + +static void write_proxy_response_if_needed(tcp::socket& socket, const protobuf::Message& request); + static protobuf::Message close_tunnel_response(const std::string& tunnel_id) { protobuf::Message response; @@ -1096,6 +1123,8 @@ static void check_client_heartbeat_and_unexpected_close_response() auto heartbeat = read_message(socket); assert(heartbeat.has_heartbeat()); + assert(heartbeat.heartbeat().sequence() == 0); + assert(heartbeat.heartbeat().acknowledgement() == 0); write_message(socket, close_control_response()); wait_for_peer_close(socket); }); @@ -1104,7 +1133,7 @@ static void check_client_heartbeat_and_unexpected_close_response() rstream::io_rstrm::config_client config; config.m_no_token = true; config.m_hearbeat = true; - config.m_heartbeat_interval_ms = 1; + config.m_heartbeat_interval_ms = 1000; config.m_connection_timeout_ms = kControlChannelTimeoutMs; rstream::io_rstrm::client client(io_context.get_executor(), config); @@ -1135,6 +1164,542 @@ static void check_client_heartbeat_and_unexpected_close_response() assert(disconnected); } +static void check_client_negotiates_liveness_and_accepts_delayed_acknowledgement() +{ + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + const auto& request = open_request.open_control_channel_req(); + check(request.has_liveness(), "control channel request did not advertise liveness"); + check(request.liveness().heartbeat_interval_ms() == 1000, "control channel request advertised the wrong heartbeat interval"); + check(request.liveness().heartbeat_timeout_ms() == 0, "control channel request selected its own timeout"); + write_message(socket, open_control_liveness_response(1000, 1000)); + const auto heartbeat = read_message(socket); + check(heartbeat.has_heartbeat(), "client did not send a negotiated heartbeat"); + check(heartbeat.heartbeat().sequence() == 1, "negotiated heartbeat sequence is invalid"); + check(heartbeat.heartbeat().acknowledgement() == 0, "client sent a heartbeat acknowledgement"); + std::this_thread::sleep_for(std::chrono::milliseconds(800)); + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(heartbeat.heartbeat().sequence()); + write_message(socket, acknowledgement); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + write_message(socket, close_control_response()); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool connected = false; + bool disconnected = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::server_error, "delayed heartbeat acknowledgement did not preserve the connection"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + connected = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "negotiated liveness check timed out"); + check(connected, "client did not connect with negotiated liveness"); + check(disconnected, "client did not process the post-acknowledgement close response"); +} + +static void check_client_expires_missing_liveness_acknowledgement() +{ + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + check(open_request.open_control_channel_req().has_liveness(), "control channel request did not advertise liveness"); + write_message(socket, open_control_liveness_response(1000, 1000)); + const auto heartbeat = read_message(socket); + check(heartbeat.has_heartbeat() && heartbeat.heartbeat().sequence() == 1, "client did not send the first negotiated heartbeat"); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool connected = false; + bool disconnected = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::operation_timeout, "missing heartbeat acknowledgement returned the wrong error"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + connected = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "missing liveness acknowledgement check timed out"); + check(connected, "client did not connect before liveness expiry"); + check(disconnected, "client remained connected without heartbeat acknowledgements"); +} + +static void check_client_tolerates_intermittent_heartbeat_loss() +{ + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + check(open_request.open_control_channel_req().has_liveness(), "control channel request did not advertise liveness"); + write_message(socket, open_control_liveness_response(1000, 2500)); + for (std::uint64_t sequence = 1; sequence <= 4; ++sequence) { + const auto heartbeat = read_message(socket); + check(heartbeat.has_heartbeat(), "client did not send a negotiated heartbeat"); + check(heartbeat.heartbeat().sequence() == sequence, "negotiated heartbeat sequence is invalid"); + if (sequence % 2 == 0) { + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(sequence); + write_message(socket, acknowledgement); + } + } + write_message(socket, close_control_response()); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool connected = false; + bool disconnected = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::server_error, "intermittent heartbeat loss closed the connection before recovery"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + connected = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "intermittent liveness check timed out"); + check(connected, "client did not connect before intermittent heartbeat loss"); + check(disconnected, "client did not process the post-recovery close response"); +} + +static void check_established_stream_survives_control_liveness_timeout() +{ + constexpr std::string_view request_payload = "after-control-loss"; + constexpr std::string_view reply_payload = "stream-still-alive"; + fake_engine engine; + engine.start_multi([request_payload, reply_payload](tcp::acceptor& acceptor) { + auto control = accept_connection(acceptor); + auto open_request = read_message(control); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + check(open_request.open_control_channel_req().has_liveness(), "control channel request did not advertise liveness"); + write_message(control, open_control_liveness_response(1000, 2500)); + bool tunnel_opened = false; + while (!tunnel_opened) { + const auto message = read_message(control); + if (message.has_heartbeat()) { + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(message.heartbeat().sequence()); + write_message(control, acknowledgement); + } + else { + check(message.has_open_tunnel_req(), "unexpected message before tunnel creation"); + write_message(control, open_tunnel_response(message.open_tunnel_req())); + tunnel_opened = true; + } + } + write_message(control, proxy_connection_request("stream-after-control-loss", "tunnel-1")); + auto stream = accept_connection(acceptor); + auto handshake = read_message(stream); + check(handshake.has_proxy_req(), "missing proxy stream handshake"); + check(handshake.proxy_req().stream_id() == "stream-after-control-loss", "proxy stream handshake used the wrong stream ID"); + write_proxy_response_if_needed(stream, handshake); + bool stream_acknowledged = false; + while (!stream_acknowledged) { + const auto message = read_message(control); + if (message.has_heartbeat()) { + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(message.heartbeat().sequence()); + write_message(control, acknowledgement); + } + else { + check(message.has_proxy_conn_rsp(), "unexpected message before proxy stream acknowledgement"); + check(!message.proxy_conn_rsp().has_error(), "proxy stream was rejected"); + stream_acknowledged = true; + } + } + wait_for_peer_close(control); + boost::asio::write(stream, boost::asio::buffer(std::string(request_payload))); + std::vector reply(reply_payload.size()); + boost::asio::read(stream, boost::asio::buffer(reply)); + check(std::string(reply.data(), reply.size()) == reply_payload, "established stream stopped after control liveness timeout"); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + auto tunnel = std::make_shared(); + auto accepted = std::make_shared(io_context.get_executor()); + auto endpoint = std::make_shared(); + auto request_buffer = std::make_shared>(request_payload.size()); + bool disconnected = false; + bool stream_replied = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::operation_timeout, "control liveness timeout returned the wrong error"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + rstream::io_rstrm::tunnel_properties properties; + properties.m_name = "api"; + properties.m_type = "bytestream"; + client.async_create_tunnel(properties, [&, tunnel, accepted, endpoint, request_buffer](const boost::system::error_code& create_error, rstream::io_rstrm::tunnel created_tunnel) { + check(!create_error, "failed to create tunnel before liveness timeout"); + *tunnel = std::move(created_tunnel); + tunnel->async_accept(*accepted, *endpoint, [&, accepted, request_buffer](const boost::system::error_code& accept_error) { + check(!accept_error, "failed to accept stream before liveness timeout"); + boost::asio::async_read(*accepted, boost::asio::buffer(*request_buffer), [&, accepted, request_buffer](const boost::system::error_code& read_error, std::size_t read) { + check(!read_error, "established stream read failed after control liveness timeout"); + check(read == request_buffer->size(), "established stream read was truncated after control liveness timeout"); + check(std::string(request_buffer->data(), request_buffer->size()) == request_payload, "established stream payload changed after control liveness timeout"); + auto reply = std::make_shared(reply_payload); + boost::asio::async_write(*accepted, boost::asio::buffer(*reply), [&, accepted, reply](const boost::system::error_code& write_error, std::size_t written) { + check(!write_error, "established stream write failed after control liveness timeout"); + check(written == reply->size(), "established stream reply was truncated after control liveness timeout"); + stream_replied = true; + boost::system::error_code close_error; + accepted->close(close_error); + check(!close_error, "failed to close established stream after liveness test"); + }); + }); + }); + }); + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "established stream liveness check timed out"); + check(disconnected, "control channel did not expire after missing heartbeat acknowledgements"); + check(stream_replied, "established stream did not survive control liveness timeout"); +} + +static void check_acceptor_reconnects_after_liveness_timeout_without_breaking_established_stream() +{ + constexpr std::string_view request_payload = "after-acceptor-reconnect"; + constexpr std::string_view reply_payload = "accepted-stream-survived"; + std::atomic_bool second_tunnel_online = false; + fake_engine engine; + engine.start_multi([request_payload, reply_payload, &second_tunnel_online](tcp::acceptor& network_acceptor) { + auto first_control = accept_connection(network_acceptor); + auto open_request = read_message(first_control); + check(open_request.has_open_control_channel_req(), "missing first acceptor control request"); + write_message(first_control, open_control_liveness_response(1000, 2500)); + auto tunnel_request = read_control_message_and_acknowledge_heartbeats(first_control); + check(tunnel_request.has_open_tunnel_req(), "missing first acceptor tunnel request"); + write_message(first_control, open_tunnel_response(tunnel_request.open_tunnel_req())); + write_message(first_control, proxy_connection_request("accepted-before-control-loss", "tunnel-1")); + auto stream = accept_connection(network_acceptor); + auto handshake = read_message(stream); + check(handshake.has_proxy_req(), "missing accepted-stream handshake"); + check(handshake.proxy_req().stream_id() == "accepted-before-control-loss", "accepted-stream handshake used the wrong stream ID"); + write_proxy_response_if_needed(stream, handshake); + auto stream_acknowledgement = read_control_message_and_acknowledge_heartbeats(first_control); + check(stream_acknowledgement.has_proxy_conn_rsp(), "missing accepted-stream acknowledgement"); + check(!stream_acknowledgement.proxy_conn_rsp().has_error(), "acceptor rejected the established stream"); + wait_for_peer_close(first_control); + auto second_control = accept_connection(network_acceptor); + open_request = read_message(second_control); + check(open_request.has_open_control_channel_req(), "acceptor did not reconnect after liveness timeout"); + write_message(second_control, open_control_liveness_response(1000, 2500)); + tunnel_request = read_control_message_and_acknowledge_heartbeats(second_control); + check(tunnel_request.has_open_tunnel_req(), "acceptor did not recreate its tunnel after reconnecting"); + write_message(second_control, open_tunnel_response(tunnel_request.open_tunnel_req())); + wait_until_ready(second_tunnel_online, "acceptor did not report its recreated tunnel online"); + boost::asio::write(stream, boost::asio::buffer(std::string(request_payload))); + std::vector reply(reply_payload.size()); + boost::asio::read(stream, boost::asio::buffer(reply)); + check(std::string(reply.data(), reply.size()) == reply_payload, "accepted stream stopped after acceptor reconnection"); + auto close_request = read_control_message_and_acknowledge_heartbeats(second_control); + check(close_request.has_close_control_channel_req(), "acceptor did not close the reconnected control channel cleanly"); + write_message(second_control, close_control_response()); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::settings_acceptor settings; + settings.m_config.m_no_token = true; + settings.m_config.m_hearbeat = true; + settings.m_config.m_heartbeat_interval_ms = 1000; + settings.m_config.m_connection_timeout_ms = kControlChannelTimeoutMs; + settings.m_auto_reconnect = true; + settings.m_reconnect_timeout_ms = 1; + settings.m_auto_recreate_tunnel = true; + settings.m_recreate_tunnel_timeout_ms = 1; + settings.m_tunnel_properties.m_type = "bytestream"; + settings.m_tunnel_properties.m_publish = true; + rstream::io_rstrm::acceptor network_acceptor(io_context.get_executor(), settings); + rstream::io_rstrm::endpoint endpoint; + endpoint.m_id_name = "api"; + endpoint.m_server_address = rstream::io::make_address(engine.address()); + boost::system::error_code bind_error; + network_acceptor.bind(endpoint, bind_error); + check(!bind_error, "failed to bind reconnecting acceptor"); + std::size_t online_status_count = 0; + std::size_t disconnected_count = 0; + rstream::io_rstrm::acceptor::control_callbacks callbacks; + callbacks.m_on_status_cb = [&](const rstream::io_rstrm::status_extd& status) { + if (status.m_status && status.m_status.value() == "online") { + ++online_status_count; + if (online_status_count == 2) { + second_tunnel_online.store(true, std::memory_order_release); + } + } + else if (status.m_status && status.m_status.value().starts_with("disconnected")) { + ++disconnected_count; + } + }; + boost::system::error_code callback_error; + network_acceptor.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install reconnecting acceptor callbacks"); + rstream::io_rstrm::socket peer(io_context.get_executor()); + rstream::io_rstrm::endpoint accepted_endpoint; + auto request = std::make_shared>(request_payload.size()); + bool stream_replied = false; + watchdog test_watchdog(io_context); + network_acceptor.async_accept(peer, accepted_endpoint, [&](const boost::system::error_code& accept_error) { + check(!accept_error, "acceptor failed to establish the stream before control loss"); + boost::asio::async_read(peer, boost::asio::buffer(*request), [&](const boost::system::error_code& read_error, std::size_t read) { + check(!read_error, "accepted stream read failed after acceptor reconnection"); + check(read == request->size(), "accepted stream read was truncated after acceptor reconnection"); + check(std::string(request->data(), request->size()) == request_payload, "accepted stream payload changed after acceptor reconnection"); + auto reply = std::make_shared(reply_payload); + boost::asio::async_write(peer, boost::asio::buffer(*reply), [&, reply](const boost::system::error_code& write_error, std::size_t written) { + check(!write_error, "accepted stream write failed after acceptor reconnection"); + check(written == reply->size(), "accepted stream reply was truncated after acceptor reconnection"); + stream_replied = true; + boost::system::error_code close_error; + peer.close(close_error); + check(!close_error, "failed to close accepted stream after reconnect test"); + network_acceptor.close(close_error); + check(!close_error, "failed to close reconnecting acceptor"); + }); + }); + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "acceptor reconnect check timed out"); + check(disconnected_count == 1, "acceptor reported " + std::to_string(disconnected_count) + " liveness disconnections instead of one"); + check(online_status_count == 2, "acceptor reported " + std::to_string(online_status_count) + " online states instead of two"); + check(stream_replied, "accepted stream did not survive acceptor reconnection"); +} + +static void check_client_rejects_invalid_liveness_acknowledgement() +{ + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + check(open_request.open_control_channel_req().has_liveness(), "control channel request did not advertise liveness"); + write_message(socket, open_control_liveness_response(1000, 60000)); + const auto heartbeat = read_message(socket); + check(heartbeat.has_heartbeat() && heartbeat.heartbeat().sequence() == 1, "client did not send the first negotiated heartbeat"); + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(heartbeat.heartbeat().sequence() + 1); + write_message(socket, acknowledgement); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool connected = false; + bool disconnected = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::protocol_error, "invalid heartbeat acknowledgement returned the wrong error"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + connected = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "invalid liveness acknowledgement check timed out"); + check(connected, "client did not connect before invalid acknowledgement"); + check(disconnected, "client accepted an invalid heartbeat acknowledgement"); +} + +static void check_client_rejects_replayed_liveness_acknowledgement() +{ + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + write_message(socket, open_control_liveness_response(1000, 60000)); + const auto heartbeat = read_message(socket); + check(heartbeat.has_heartbeat() && heartbeat.heartbeat().sequence() == 1, "client did not send the first negotiated heartbeat"); + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(heartbeat.heartbeat().sequence()); + write_message(socket, acknowledgement); + write_message(socket, acknowledgement); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool connected = false; + bool disconnected = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::protocol_error, "replayed heartbeat acknowledgement returned the wrong error"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + connected = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "replayed liveness acknowledgement check timed out"); + check(connected, "client did not connect before replayed acknowledgement"); + check(disconnected, "client accepted a replayed heartbeat acknowledgement"); +} + +static void check_client_rejects_invalid_liveness_configuration() +{ + for (const auto interval : {999U, 300001U}) { + fake_engine engine; + engine.start([](tcp::socket& socket) { wait_for_peer_close(socket); }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = interval; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool completed = false; + watchdog test_watchdog(io_context); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::invalid_configuration, "invalid heartbeat interval returned the wrong error"); + completed = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "invalid heartbeat interval check timed out"); + check(completed, "invalid heartbeat interval did not complete"); + } +} + +static void check_client_rejects_invalid_server_liveness_policy() +{ + const std::array, 3> policies = {{{2000, 60000}, {1000, 999}, {1000, 900001}}}; + for (const auto& policy : policies) { + fake_engine engine; + engine.start([policy](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + write_message(socket, open_control_liveness_response(policy.first, policy.second)); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool completed = false; + watchdog test_watchdog(io_context); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::protocol_error, "invalid server liveness policy returned the wrong error"); + completed = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "invalid server liveness policy check timed out"); + check(completed, "invalid server liveness policy did not complete"); + } + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + check(!open_request.open_control_channel_req().has_liveness(), "disabled heartbeat advertised liveness"); + write_message(socket, open_control_liveness_response(1000, 60000)); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = false; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool completed = false; + watchdog test_watchdog(io_context); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::protocol_error, "unsolicited server liveness policy returned the wrong error"); + completed = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "unsolicited server liveness policy check timed out"); + check(completed, "unsolicited server liveness policy did not complete"); +} + static void check_client_can_reconnect_from_disconnection_callback() { fake_engine first_engine; @@ -1626,6 +2191,70 @@ static void check_socket_rejects_invalid_state_operations() assert(!close_error); } +static void check_error_completions_are_thread_safe() +{ + constexpr std::size_t operation_count = 256; + boost::asio::io_context io_context; + std::vector> sockets; + std::vector> clients; + std::vector> acceptors; + std::vector> accepted_peers; + std::vector> accepted_endpoints; + sockets.reserve(operation_count); + clients.reserve(operation_count); + acceptors.reserve(operation_count); + accepted_peers.reserve(operation_count); + accepted_endpoints.reserve(operation_count); + std::atomic_size_t completions = 0; + std::atomic_size_t failures = 0; + for (std::size_t index = 0; index < operation_count; ++index) { + sockets.push_back(std::make_unique(io_context.get_executor())); + rstream::io_rstrm::endpoint missing_id; + missing_id.m_server_address = rstream::io::make_address("127.0.0.1:1"); + sockets.back()->async_connect(missing_id, [&](const boost::system::error_code& error_code) { + if (error_code != rstream::io_rstrm::error::code::invalid_endpoint) { + failures.fetch_add(1, std::memory_order_relaxed); + } + completions.fetch_add(1, std::memory_order_relaxed); + }); + + rstream::io_rstrm::config_client client_config; + client_config.m_no_token = true; + clients.push_back(std::make_unique(io_context.get_executor(), client_config)); + rstream::io_rstrm::tunnel_properties properties; + clients.back()->async_create_tunnel(properties, [&](const boost::system::error_code& error_code, rstream::io_rstrm::tunnel) { + if (error_code != rstream::io_rstrm::error::code::invalid_state) { + failures.fetch_add(1, std::memory_order_relaxed); + } + completions.fetch_add(1, std::memory_order_relaxed); + }); + + rstream::io_rstrm::settings_acceptor acceptor_settings; + acceptor_settings.m_config.m_no_token = true; + acceptors.push_back(std::make_unique(io_context.get_executor(), acceptor_settings)); + accepted_peers.push_back(std::make_unique(io_context.get_executor())); + accepted_endpoints.push_back(std::make_unique()); + acceptors.back()->async_accept( + *accepted_peers.back(), + *accepted_endpoints.back(), + [&](const boost::system::error_code& error_code) { + if (error_code != rstream::io_rstrm::error::code::no_valid_endpoint) { + failures.fetch_add(1, std::memory_order_relaxed); + } + completions.fetch_add(1, std::memory_order_relaxed); + }); + } + std::array workers; + for (auto& worker : workers) { + worker = std::thread([&] { io_context.run(); }); + } + for (auto& worker : workers) { + worker.join(); + } + assert(failures.load(std::memory_order_relaxed) == 0); + assert(completions.load(std::memory_order_relaxed) == operation_count * 3); +} + static void check_async_connect_freezes_client_configuration() { boost::asio::io_context io_context; @@ -2457,12 +3086,22 @@ int main(int argc, char** argv) run_selected_check(selected, "client_rejects_malformed_tunnel_responses", check_client_rejects_malformed_tunnel_responses); run_selected_check(selected, "client_rejects_duplicate_active_tunnel_id", check_client_rejects_duplicate_active_tunnel_id); run_selected_check(selected, "client_heartbeat_and_unexpected_close_response", check_client_heartbeat_and_unexpected_close_response); + run_selected_check(selected, "client_negotiates_liveness_and_accepts_delayed_acknowledgement", check_client_negotiates_liveness_and_accepts_delayed_acknowledgement); + run_selected_check(selected, "client_expires_missing_liveness_acknowledgement", check_client_expires_missing_liveness_acknowledgement); + run_selected_check(selected, "client_tolerates_intermittent_heartbeat_loss", check_client_tolerates_intermittent_heartbeat_loss); + run_selected_check(selected, "established_stream_survives_control_liveness_timeout", check_established_stream_survives_control_liveness_timeout); + run_selected_check(selected, "acceptor_reconnects_after_liveness_timeout_without_breaking_established_stream", check_acceptor_reconnects_after_liveness_timeout_without_breaking_established_stream); + run_selected_check(selected, "client_rejects_invalid_liveness_acknowledgement", check_client_rejects_invalid_liveness_acknowledgement); + run_selected_check(selected, "client_rejects_replayed_liveness_acknowledgement", check_client_rejects_replayed_liveness_acknowledgement); + run_selected_check(selected, "client_rejects_invalid_liveness_configuration", check_client_rejects_invalid_liveness_configuration); + run_selected_check(selected, "client_rejects_invalid_server_liveness_policy", check_client_rejects_invalid_server_liveness_policy); run_selected_check(selected, "client_can_reconnect_from_disconnection_callback", check_client_can_reconnect_from_disconnection_callback); run_selected_check(selected, "client_drains_active_control_write_before_reconnect", check_client_drains_active_control_write_before_reconnect); run_selected_check(selected, "client_accepts_delayed_proxy_stream_and_rejects_max_streams", check_client_accepts_delayed_proxy_stream_and_rejects_max_streams); run_selected_check(selected, "client_rejects_proxy_request_for_unknown_tunnel", check_client_rejects_proxy_request_for_unknown_tunnel); run_selected_check(selected, "client_reports_redirected_proxy_failures", check_client_reports_redirected_proxy_failures); run_selected_check(selected, "socket_rejects_invalid_state_operations", check_socket_rejects_invalid_state_operations); + run_selected_check(selected, "error_completions_are_thread_safe", check_error_completions_are_thread_safe); run_selected_check(selected, "async_connect_freezes_client_configuration", check_async_connect_freezes_client_configuration); run_selected_check(selected, "async_connect_freezes_socket_configuration", check_async_connect_freezes_socket_configuration); run_selected_check(selected, "async_accept_freezes_acceptor_configuration", check_async_accept_freezes_acceptor_configuration); diff --git a/test/io/io-rstrm/test_io_rstrm_handshake.cpp b/test/io/io-rstrm/test_io_rstrm_handshake.cpp index dabe241..f575b0c 100644 --- a/test/io/io-rstrm/test_io_rstrm_handshake.cpp +++ b/test/io/io-rstrm/test_io_rstrm_handshake.cpp @@ -1,10 +1,14 @@ // See LICENSE file in the project root for license information. +#include #include #include #include +#include #include +#include #include +#include #include #include @@ -13,7 +17,9 @@ #include #include #include +#include #include +#include #include #include @@ -103,7 +109,11 @@ static void check_zero_rtt_stream_request_does_not_wait_for_response() rstream::test::connect_stream_pair(*socket_a, *socket_b); test_stream stream(*socket_a, true); rstream::io_rstrm::config config; +#ifdef RSTREAM_WITH_IO_STREAMS config.m_token = "secret-token"; +#else + config.m_no_token = true; +#endif config.m_zero_rtt = true; bool client_called = false; bool server_called = false; @@ -123,8 +133,12 @@ static void check_zero_rtt_stream_request_does_not_wait_for_response() assert(message.stream_req().tunnel_id_name() == "api"); assert(message.stream_req().has_zero_rtt()); assert(message.stream_req().zero_rtt().value()); +#ifdef RSTREAM_WITH_IO_STREAMS assert(message.stream_req().client_details().has_token()); assert(message.stream_req().client_details().token().value() == "secret-token"); +#else + assert(!message.stream_req().client_details().has_token()); +#endif server_called = true; socket->close(); co_return; }, boost::asio::detached); @@ -256,6 +270,54 @@ static void check_proxy_success_response_completes() assert(server_called); } +static void check_proxy_response_preserves_coalesced_payload() +{ + boost::asio::io_context io_context; + auto socket_a = std::make_shared(io_context.get_executor()); + auto socket_b = std::make_shared(io_context.get_executor()); + rstream::test::connect_stream_pair(*socket_a, *socket_b); + test_stream stream(*socket_a, false); + rstream::io_rstrm::config config; + config.m_no_token = true; + config.m_zero_rtt = false; + bool handshake_called = false; + bool payload_called = false; + std::array received{}; + handshake_type handshake(stream, rstream::io::make_address("engine.example:443"), config); + handshake.async_run(handshake_type::type::proxy_req, "stream-123", boost::none, [&](const boost::system::error_code& error_code) { + handshake_called = true; + assert(!error_code); + boost::asio::async_read(*socket_a, boost::asio::buffer(received), [&](const boost::system::error_code& read_error, std::size_t size) { + assert(!read_error); + assert(size == received.size()); + payload_called = true; + }); + }); + boost::asio::co_spawn(io_context.get_executor(), [socket = socket_b]() -> boost::asio::awaitable { + payloader_type payloader(*socket); + auto request = rstream::core::make_buffer_allocated(4096); + co_await payloader.async_recv(request, boost::asio::use_awaitable); + protobuf::Message response; + response.mutable_proxy_rsp(); + auto payload = serialize_message(response); + const std::string application_payload = "world"; + const auto payload_size = static_cast(payload.get_size()); + std::vector wire(sizeof(payload_size) + payload_size + application_payload.size()); + wire[0] = static_cast(payload_size >> 24); + wire[1] = static_cast(payload_size >> 16); + wire[2] = static_cast(payload_size >> 8); + wire[3] = static_cast(payload_size); + std::memcpy(wire.data() + sizeof(payload_size), payload.map().get_const_data(), payload_size); + std::memcpy(wire.data() + sizeof(payload_size) + payload_size, application_payload.data(), application_payload.size()); + co_await boost::asio::async_write(*socket, boost::asio::buffer(wire), boost::asio::use_awaitable); + co_return; }, boost::asio::detached); + io_context.run(); + assert(handshake_called); + assert(payload_called); + assert(std::string(received.data(), received.size()) == "world"); +} + +#ifdef RSTREAM_WITH_IO_STREAMS static void check_proxy_secret_is_allowed_with_mtls_agent_auth() { boost::asio::io_context io_context; @@ -293,6 +355,7 @@ static void check_proxy_secret_is_allowed_with_mtls_agent_auth() assert(client_called); assert(server_called); } +#endif static void check_unexpected_response_type_is_rejected() { @@ -429,7 +492,10 @@ int main(int argc, char** argv) check_stream_response_error_is_mapped(); check_stream_success_response_completes(); check_proxy_success_response_completes(); + check_proxy_response_preserves_coalesced_payload(); +#ifdef RSTREAM_WITH_IO_STREAMS check_proxy_secret_is_allowed_with_mtls_agent_auth(); +#endif check_unexpected_response_type_is_rejected(); check_invalid_protobuf_response_is_rejected(); check_cancellation_reaches_the_transport(); diff --git a/test/ncat/CMakeLists.txt b/test/ncat/CMakeLists.txt index cc0d3c0..2a9290b 100644 --- a/test/ncat/CMakeLists.txt +++ b/test/ncat/CMakeLists.txt @@ -20,4 +20,9 @@ if(WIN32) else() add_test_target(${PROJECT_NAME}-test-ncat-runtime test_ncat_runtime.cpp ${PROJECT_NAME}::ncat ${PROJECT_NAME}::${PROJECT_NAME}) rstream_enable_runtime_plugins(${PROJECT_NAME}-test-ncat-runtime ${PROJECT_NAME}-plugin-io-generic) + add_test( + NAME ${PROJECT_NAME}-test-ncat-cli-runtime + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test_ncat_cli_runtime.py $) + rstream_configure_test(${PROJECT_NAME}-test-ncat-cli-runtime) + set_property(GLOBAL APPEND PROPERTY RSTREAM_TEST_TARGETS ${PROJECT_NAME}-ncat) endif() diff --git a/test/ncat/test_ncat_cli_runtime.py b/test/ncat/test_ncat_cli_runtime.py new file mode 100644 index 0000000..9ce4066 --- /dev/null +++ b/test/ncat/test_ncat_cli_runtime.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python3 + +import signal +import socket +import subprocess +import sys +import time + + +def allocate_address() -> str: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener: + listener.bind(("127.0.0.1", 0)) + return f"127.0.0.1:{listener.getsockname()[1]}" + + +def run_case(binary: str, command_arguments: list[str], expected: bytes) -> None: + address = allocate_address() + server = subprocess.Popen( + [binary, "-L", address, *command_arguments, "--jobs=2"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + try: + deadline = time.monotonic() + 10 + while True: + client = subprocess.run( + [binary, address, "-I", "--jobs=2"], + capture_output=True, + timeout=10, + check=False, + ) + if client.returncode == 0: + if client.stdout != expected: + raise AssertionError( + f"unexpected output {client.stdout!r}, expected {expected!r}; " + f"client stderr={client.stderr!r}" + ) + break + if server.poll() is not None: + stdout, stderr = server.communicate() + raise AssertionError( + f"server exited with {server.returncode}; stdout={stdout!r}; stderr={stderr!r}" + ) + if time.monotonic() >= deadline: + raise AssertionError(f"client did not connect; stderr={client.stderr!r}") + time.sleep(0.05) + finally: + if server.poll() is None: + server.send_signal(signal.SIGTERM) + try: + server.communicate(timeout=10) + except subprocess.TimeoutExpired: + server.kill() + server.communicate() + + +def main() -> None: + if len(sys.argv) != 2: + raise SystemExit("usage: test_ncat_cli_runtime.py ") + command = "exec printf cli-shell-ok" + run_case(sys.argv[1], ["-c", command], b"cli-shell-ok") + run_case(sys.argv[1], [f"-c={command}"], b"cli-shell-ok") + + +if __name__ == "__main__": + main() diff --git a/test/ncat/test_ncat_runtime.cpp b/test/ncat/test_ncat_runtime.cpp index 0ed1b86..c7fd229 100644 --- a/test/ncat/test_ncat_runtime.cpp +++ b/test/ncat/test_ncat_runtime.cpp @@ -2,7 +2,9 @@ #include #include +#include #include +#include #include #include #include @@ -19,6 +21,7 @@ #include #include +#include #include #include @@ -510,6 +513,46 @@ static void check_exec_server_pipes_downstream_to_child() assert(response.find("exec") != std::string::npos); } +static int run_exec_server_write_after_child_closed_stdin() +{ + std::signal(SIGPIPE, SIG_DFL); + const auto port = unused_tcp_port(); + rstream::ncat::server::config config = { + .m_local = rstream::io::address(std::string("127.0.0.1:") + std::to_string(port)), + .m_remote = rstream::ncat::server::exec{.m_shell = true, .m_cmd = "exec 0<&-; printf ready; sleep 30"}, + }; + ncat_server_fixture server(config); + server.start(); + boost::asio::io_context io_context; + auto socket = connect_with_retry(io_context, server.port()); + std::array ready{}; + boost::asio::read(socket, boost::asio::buffer(ready)); + assert(std::string(ready.data(), ready.size()) == "ready"); + boost::system::error_code error_code; + boost::asio::write(socket, boost::asio::buffer(std::string("payload")), error_code); + std::array response{}; + socket.read_some(boost::asio::buffer(response), error_code); + assert(error_code); + server.stop(); + return 0; +} + +static void check_exec_server_write_after_child_closed_stdin_does_not_raise_sigpipe(const char* executable) +{ + const auto pid = ::fork(); + require_posix(pid != -1, "fork failed"); + if (pid == 0) { + ::execl(executable, executable, "--exec-write-after-child-closed-stdin", static_cast(nullptr)); + ::_exit(127); + } + int status = 0; + while (::waitpid(pid, &status, 0) == -1) { + require_posix(errno == EINTR, "waitpid failed"); + } + assert(WIFEXITED(status)); + assert(WEXITSTATUS(status) == 0); +} + static void check_proxy_server_pipes_downstream_to_upstream() { echo_upstream upstream; @@ -685,8 +728,10 @@ static void check_exec_server_cancel_keeps_active_child_alive_until_completion() int main(int argc, char** argv) { - (void)argc; - (void)argv; + if (argc == 2 && std::string(argv[1]) == "--exec-write-after-child-closed-stdin") { + return run_exec_server_write_after_child_closed_stdin(); + } + check_exec_server_write_after_child_closed_stdin_does_not_raise_sigpipe(argv[0]); check_error_category(); check_exec_server_pipes_downstream_to_child(); check_proxy_server_pipes_downstream_to_upstream(); diff --git a/test/tunnel/test_tunnel_proxy.cpp b/test/tunnel/test_tunnel_proxy.cpp index abe490d..e53e690 100644 --- a/test/tunnel/test_tunnel_proxy.cpp +++ b/test/tunnel/test_tunnel_proxy.cpp @@ -222,9 +222,10 @@ class tcp_thread_server { class fake_engine { public: - explicit fake_engine(std::atomic_bool& stream_exchanged) + explicit fake_engine(std::atomic_bool& stream_exchanged, bool half_close = false) : m_acceptor(m_io_context, tcp::endpoint(boost::asio::ip::make_address("127.0.0.1"), 0)), - m_stream_exchanged(stream_exchanged) + m_stream_exchanged(stream_exchanged), + m_half_close(half_close) { } @@ -272,6 +273,11 @@ class fake_engine { const std::string downstream_payload = "hello"; boost::asio::write(stream, boost::asio::buffer(downstream_payload)); + if (m_half_close) { + boost::system::error_code error_code; + stream.shutdown(tcp::socket::shutdown_send, error_code); + check(!error_code, "failed to half-close downstream test stream"); + } std::array upstream_reply{}; boost::asio::read(stream, boost::asio::buffer(upstream_reply)); check(std::string(upstream_reply.data(), upstream_reply.size()) == "world", "unexpected proxied reply"); @@ -301,25 +307,37 @@ class fake_engine { boost::asio::io_context m_io_context; tcp::acceptor m_acceptor; std::atomic_bool& m_stream_exchanged; + bool m_half_close; std::thread m_thread; std::exception_ptr m_exception; }; -static void check_proxy_forwards_engine_stream_to_upstream_and_back() +static void check_proxy_forwards_engine_stream_to_upstream_and_back(bool half_close = false) { std::atomic_bool upstream_served = false; - tcp_thread_server upstream([&](tcp::socket& socket) { + tcp_thread_server upstream([&, half_close](tcp::socket& socket) { std::array request{}; boost::asio::read(socket, boost::asio::buffer(request)); check(std::string(request.data(), request.size()) == "hello", "unexpected upstream request"); + if (half_close) { + std::array trailing{}; + boost::system::error_code error_code; + socket.read_some(boost::asio::buffer(trailing), error_code); + check(error_code == boost::asio::error::eof, "upstream did not receive the downstream half-close"); + } const std::string response = "world"; boost::asio::write(socket, boost::asio::buffer(response)); + if (half_close) { + boost::system::error_code error_code; + socket.shutdown(tcp::socket::shutdown_send, error_code); + check(!error_code, "failed to half-close upstream test stream"); + } upstream_served = true; }); upstream.start(); std::atomic_bool stream_exchanged = false; - fake_engine engine(stream_exchanged); + fake_engine engine(stream_exchanged, half_close); engine.start(); boost::asio::io_context io_context; @@ -500,6 +518,7 @@ int main(int argc, char** argv) (void)argc; (void)argv; check_proxy_forwards_engine_stream_to_upstream_and_back(); + check_proxy_forwards_engine_stream_to_upstream_and_back(true); check_proxy_rejects_second_run_while_active(); check_proxy_default_tunnel_request_leaves_public_policy_to_server(); return 0; diff --git a/test/webtty/test_webtty_plain_client_runtime.cpp b/test/webtty/test_webtty_plain_client_runtime.cpp index 608c47b..c2ad823 100644 --- a/test/webtty/test_webtty_plain_client_runtime.cpp +++ b/test/webtty/test_webtty_plain_client_runtime.cpp @@ -1,5 +1,6 @@ // See LICENSE file in the project root for license information. +#include #include #include #include @@ -516,6 +517,13 @@ class fake_early_exit_plain_server { protobuf::Message close; close.mutable_close()->set_return_code(0); write_message(socket, close); + boost::system::error_code error_code; + socket.shutdown(tcp::socket::shutdown_send, error_code); + assert(!error_code); + std::array drain_buffer{}; + while (socket.read_some(boost::asio::buffer(drain_buffer), error_code) != 0) { + } + assert(error_code == boost::asio::error::eof); } catch (...) { m_exception = std::current_exception(); diff --git a/test/webtty/test_webtty_plain_server_runtime.cpp b/test/webtty/test_webtty_plain_server_runtime.cpp index 7ce52d5..3e51e8b 100644 --- a/test/webtty/test_webtty_plain_server_runtime.cpp +++ b/test/webtty/test_webtty_plain_server_runtime.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include #include @@ -63,6 +64,16 @@ static void write_message(tcp::socket& socket, const protobuf::Message& message) } } +static void append_message(std::vector& dst, const protobuf::Message& message) +{ + const auto size = static_cast(message.ByteSizeLong()); + std::uint32_t frame_size = htonl(size); + const auto offset = dst.size(); + dst.resize(offset + sizeof(frame_size) + size); + std::memcpy(dst.data() + offset, &frame_size, sizeof(frame_size)); + assert(message.SerializeToArray(dst.data() + offset + sizeof(frame_size), static_cast(size))); +} + static void read_exact(tcp::socket& socket, void* data, std::size_t size) { std::size_t offset = 0; @@ -656,6 +667,46 @@ static void check_plain_server_reports_child_exit_without_stdin_eos(unsigned sho } } +static void check_plain_server_delivers_child_exit_while_messages_are_in_flight(unsigned short port) +{ + boost::asio::io_context io_context; + auto socket = connect_with_retry(io_context, port); + write_message(socket, open_message({"/bin/sh", "-c", "IFS= read -r line; exit 23"}, true)); + + auto ack = read_message(socket); + assert(ack.payload_case() == protobuf::Message::PayloadCase::kAck); + protobuf::Message heartbeat; + heartbeat.mutable_heartbeat(); + std::vector in_flight; + in_flight.reserve(512 * 1024); + for (int i = 0; i < 1000; ++i) { + append_message(in_flight, heartbeat); + } + append_message(in_flight, data_message(protobuf::Data::TYPE_STDIN, "exit\n")); + for (int i = 0; i < 50000; ++i) { + append_message(in_flight, heartbeat); + } + boost::asio::write(socket, boost::asio::buffer(in_flight)); + + bool saw_close = false; + while (!saw_close) { + auto message = read_message(socket); + switch (message.payload_case()) { + case protobuf::Message::PayloadCase::kData: + case protobuf::Message::PayloadCase::kHeartbeat: + break; + case protobuf::Message::PayloadCase::kClose: + assert(message.close().return_code() == 23); + saw_close = true; + break; + default: + std::cerr << "unexpected webtty message type: " << message.payload_case() << std::endl; + assert(false); + break; + } + } +} + static void check_plain_server_e2e_forwards_stdin_to_child_process() { std::error_code error_code; @@ -972,9 +1023,31 @@ static void check_plain_server_cancel_keeps_active_child_resources_alive() assert(out.payload_case() == protobuf::Message::PayloadCase::kData); assert(out.data().type() == protobuf::Data::TYPE_STDOUT); assert(out.data().data() == "active"); - server.stop(); + std::exception_ptr stop_exception; + std::thread stop_thread([&] { + try { + server.stop(); + } + catch (...) { + stop_exception = std::current_exception(); + } + }); + bool saw_close = false; + while (!saw_close) { + auto message = read_message(socket); + if (message.payload_case() == protobuf::Message::PayloadCase::kClose) { + saw_close = true; + } + else { + assert(message.payload_case() == protobuf::Message::PayloadCase::kData || message.payload_case() == protobuf::Message::PayloadCase::kHeartbeat); + } + } boost::system::error_code ignored; socket.close(ignored); + stop_thread.join(); + if (stop_exception) { + std::rethrow_exception(stop_exception); + } } } @@ -1002,6 +1075,7 @@ int main(int argc, char** argv) run_check("tty environment and workdir", [&server] { check_plain_server_applies_tty_environment_and_workdir(server.port()); }); run_check("stdin forwarding", [&server] { check_plain_server_forwards_stdin_to_child_process(server.port()); }); run_check("child exit before stdin EOS", [&server] { check_plain_server_reports_child_exit_without_stdin_eos(server.port()); }); + run_check("child exit while messages are in flight", [&server] { check_plain_server_delivers_child_exit_while_messages_are_in_flight(server.port()); }); run_check("E2E stdin forwarding", check_plain_server_e2e_forwards_stdin_to_child_process); run_check("client credential verification", check_plain_server_e2e_accepts_client_credential_verifier); run_check("missing session key rejection", check_plain_server_e2e_rejects_missing_session_key_grant); diff --git a/test/webtty/test_webtty_terminal_stream.cpp b/test/webtty/test_webtty_terminal_stream.cpp index 802a12f..e0e760d 100644 --- a/test/webtty/test_webtty_terminal_stream.cpp +++ b/test/webtty/test_webtty_terminal_stream.cpp @@ -4,6 +4,7 @@ #include #include +#include #include #include #include @@ -28,6 +29,7 @@ #ifdef _WIN32 #include #else +#include #include #ifdef __APPLE__ #include @@ -240,15 +242,112 @@ static void check_pipe_stream_lifecycle() auto stream_ptr = stream::make_stream(io_context.get_executor(), stream::backend::pipe); assert(stream_ptr); assert(stream_ptr->backend() == stream::backend::pipe); - assert(std::dynamic_pointer_cast(stream_ptr)); + auto pipe = std::dynamic_pointer_cast(stream_ptr); + assert(pipe); + bool invalid_output_rejected = false; + try { + (void)pipe->output_stream(stream::type::std_in); + } + catch (const std::invalid_argument&) { + invalid_output_rejected = true; + } + assert(invalid_output_rejected); stream_ptr->close(); stream_ptr->close(); } +#ifndef _WIN32 +static int run_pipe_write_after_child_exit() +{ + std::signal(SIGPIPE, SIG_DFL); + boost::asio::io_context io_context; + auto stream_ptr = stream::make_stream(io_context.get_executor(), stream::backend::pipe); + auto child = rstream::webtty::detail::process::make_child( + stream_ptr, + boost::process::exe("/bin/sh"), + boost::process::args(std::vector{"-c", "exit 0"})); + child->wait(); + std::vector payload(1024 * 1024, 'x'); + std::error_code write_error; + std::size_t completions = 0; + stream_ptr->async_write(boost::asio::buffer(payload), stream::type::std_in, [&](const std::error_code& error_code, std::size_t) { + write_error = error_code; + ++completions; + }); + io_context.run(); + assert(completions == 1); + assert(write_error); + return 0; +} + +static void check_pipe_subprocess(const char* executable, const char* mode) +{ + boost::process::child child( + boost::process::exe(executable), + boost::process::args(std::vector{mode})); + child.wait(); + assert(child.exit_code() == 0); +} + +static void check_pipe_write_after_child_exit_does_not_raise_sigpipe(const char* executable) +{ + check_pipe_subprocess(executable, "--pipe-write-after-child-exit"); + check_pipe_subprocess(executable, "--pipe-write-after-child-exit-with-closed-stdin"); +} + +static std::size_t open_descriptor_count() +{ + auto directory = ::opendir("/dev/fd"); + assert(directory != nullptr); + std::size_t count = 0; + while (const auto* entry = ::readdir(directory)) { + if (entry->d_name[0] != '.') { + ++count; + } + } + assert(::closedir(directory) == 0); + return count; +} + +static void spawn_pipe_child_once() +{ + boost::asio::io_context io_context; + auto stream_ptr = stream::make_stream(io_context.get_executor(), stream::backend::pipe); + auto child = rstream::webtty::detail::process::make_child( + stream_ptr, + boost::process::exe("/bin/sh"), + boost::process::args(std::vector{"-c", "exit 0"})); + child->wait(); + stream_ptr->close(); +} + +static void check_pipe_child_spawn_does_not_leak_descriptors() +{ + spawn_pipe_child_once(); + const auto before = open_descriptor_count(); + for (std::size_t iteration = 0; iteration < 64; ++iteration) { + spawn_pipe_child_once(); + } + assert(open_descriptor_count() == before); +} +#endif + int main(int argc, char** argv) { +#ifndef _WIN32 + if (argc == 2 && std::string(argv[1]) == "--pipe-write-after-child-exit") { + return run_pipe_write_after_child_exit(); + } + if (argc == 2 && std::string(argv[1]) == "--pipe-write-after-child-exit-with-closed-stdin") { + assert(::close(STDIN_FILENO) == 0); + return run_pipe_write_after_child_exit(); + } + check_pipe_write_after_child_exit_does_not_raise_sigpipe(argv[0]); + check_pipe_child_spawn_does_not_leak_descriptors(); +#else (void)argc; (void)argv; +#endif check_pipe_stream_lifecycle(); #ifdef _WIN32 check_windows_pty_rejects_overlapping_writes();