From 17a621f2409058ad97b3a739e648125568c810f7 Mon Sep 17 00:00:00 2001 From: khodowany Date: Thu, 8 Oct 2026 02:12:58 -0500 Subject: [PATCH] Cleanup --- CMakeLists.txt | 2 + README.md | 2 + include/ligo/detail/lifecycle.hpp | 34 +-- src/detail/dtls.cpp | 267 ++++++++++++++++++++++ src/detail/dtls.hpp | 46 ++++ src/detail/udp_protocol.cpp | 64 ++++++ src/detail/udp_protocol.hpp | 22 ++ src/udp.cpp | 365 ++---------------------------- tests/CMakeLists.txt | 1 + tests/lifecycle.cpp | 75 ++++++ 10 files changed, 517 insertions(+), 361 deletions(-) create mode 100644 src/detail/dtls.cpp create mode 100644 src/detail/dtls.hpp create mode 100644 src/detail/udp_protocol.cpp create mode 100644 src/detail/udp_protocol.hpp create mode 100644 tests/lifecycle.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 0e7efb7..2332f27 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -23,6 +23,8 @@ add_library(ligo src/service.cpp src/tcp.cpp src/udp.cpp + src/detail/dtls.cpp + src/detail/udp_protocol.cpp ) target_include_directories(ligo PUBLIC $ diff --git a/README.md b/README.md index beb9606..d21f27a 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,8 @@ Ligo is a C++23 networking and transport library. It provides HTTP JSON calls, f Ligo uses Modulus for shared build tooling, Boost.JSON, and OpenSSL integration. Product-specific routes and workflows belong in consuming applications. +UDP reliability and socket lifecycle use separate private DTLS and wire-protocol adapters. Session-drain waits serialize timer registration and notifications on a waiter strand and retain one absolute shutdown deadline, including when notifications arrive before a wait is armed. + ## Build Code style is managed by Modulus: run `./modulus.py format` or `./modulus.py format --check`. The generated Neovim integration uses the same pinned container formatter for C++ and Python, including existing format-on-save toggles. Configure first to prepare tooling; customize `tooling.editor.clang_format` and `tooling.format.exclude` in `modulus.json`, not generated editor files. See the shared Modulus README for the compact layout and no-alias policy. diff --git a/include/ligo/detail/lifecycle.hpp b/include/ligo/detail/lifecycle.hpp index 21e72e7..6ee5f3a 100644 --- a/include/ligo/detail/lifecycle.hpp +++ b/include/ligo/detail/lifecycle.hpp @@ -9,6 +9,8 @@ #include #include +#include +#include #include #include #include @@ -40,6 +42,7 @@ struct session_registry { auto cancel_all() -> void; auto empty() const -> std::bool_t; auto wait_empty(std::chrono::milliseconds timeout) -> boost::asio::awaitable>; + auto wait_registered(std::shared_ptr timer, std::chrono::steady_clock::time_point deadline) -> boost::asio::awaitable>; auto notify_waiters() -> void; auto live_waiters_locked() -> std::vector>; mutable std::mutex mutex{}; @@ -74,7 +77,8 @@ auto ligo::detail::cancel_session_stream(boost::asio::ssl::stream &s template auto ligo::detail::session_registry::cancel_waiters(std::vector> const &live) -> void { for (auto &waiter : live) { - waiter->cancel(); + // Timer operations belong to the waiter strand, including registration. + boost::asio::post(waiter->get_executor(), [waiter] { waiter->cancel(); }); } } @@ -125,30 +129,28 @@ auto ligo::detail::session_registry::wait_empty(std::chrono::milliseconds tim if (empty()) { co_return std::expected{}; } - auto notify_timer = std::make_shared(executor); - auto deadline_timer = std::make_shared(executor); - auto timed_out = std::make_shared>(false); + auto timer = std::make_shared(boost::asio::make_strand(executor)); + auto deadline = std::chrono::steady_clock::now() + timeout; { auto lock = std::scoped_lock{mutex}; - waiters.push_back(notify_timer); + waiters.push_back(timer); } - deadline_timer->expires_after(timeout); - deadline_timer->async_wait([notify_timer, timed_out](boost::system::error_code const &error) { - if (!error) { - timed_out->store(true, std::memory_order_relaxed); - notify_timer->cancel(); - } - }); + co_return co_await boost::asio::co_spawn(timer->get_executor(), wait_registered(timer, deadline), boost::asio::use_awaitable); +} + +template +auto ligo::detail::session_registry::wait_registered(std::shared_ptr timer, std::chrono::steady_clock::time_point deadline) -> boost::asio::awaitable> { + // This entire coroutine runs on the timer strand. Notifications cannot run + // between checking session state and initiating the asynchronous wait. for (;;) { if (empty()) { - deadline_timer->cancel(); co_return std::expected{}; } - if (timed_out->load(std::memory_order_relaxed)) { + if (std::chrono::steady_clock::now() >= deadline) { co_return ligo::detail::fail_void(ligo::detail::error_timeout, "timed out waiting for sessions to drain"); } - notify_timer->expires_at(std::chrono::steady_clock::time_point::max()); - auto [error] = co_await notify_timer->async_wait(boost::asio::as_tuple(boost::asio::use_awaitable)); + timer->expires_at(deadline); + auto [error] = co_await timer->async_wait(boost::asio::as_tuple(boost::asio::use_awaitable)); if (error && error != boost::asio::error::operation_aborted) { co_return ligo::detail::fail_void(ligo::detail::network_failure(error)); } diff --git a/src/detail/dtls.cpp b/src/detail/dtls.cpp new file mode 100644 index 0000000..d4a77f6 --- /dev/null +++ b/src/detail/dtls.cpp @@ -0,0 +1,267 @@ +#include "dtls.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +namespace ligo::detail { + +constexpr auto max_datagram_size = std::size_t{65507}; +constexpr auto max_dtls_plaintext_size = std::size_t{12000}; + +auto dtls_failure(std::int32_t code, std::string message) -> modulus::error { + auto const error = ERR_get_error(); + if (error != 0) { + message += ": "; + message += ERR_error_string(error, nullptr); + } + return ligo::detail::make_error(code, std::move(message)); +} + +auto dtls_failure(std::string message) -> modulus::error { + return dtls_failure(ligo::detail::error_unavailable, std::move(message)); +} + +auto validate_security(ligo::security_config const &config, std::bool_t server) -> std::expected { + if (config.mode == ligo::security_mode::TLS) { + return ligo::detail::fail_void(ligo::unsupported_security("UDP", config.mode)); + } + if (config.mode != ligo::security_mode::DTLS) { + return {}; + } + if (config.tls_context) { + return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS does not accept a TLS context")); + } + if (server && (config.certificate_chain_file.empty() || config.private_key_file.empty())) { + return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS server requires certificate chain/private key files")); + } + if (!config.certificate_chain_file.empty() != !config.private_key_file.empty()) { + return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS certificate chain and private key must both be configured")); + } + return {}; +} + +auto load_dtls_verify_paths(SSL_CTX *context, ligo::security_config const &config) -> std::expected { + if (config.verify == ligo::verify_mode::NONE) { + return {}; + } + if (!config.ca_file.empty()) { + if (SSL_CTX_load_verify_locations(context, config.ca_file.c_str(), nullptr) != 1) { + return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS CA file")); + } + return {}; + } + if (SSL_CTX_set_default_verify_paths(context) != 1) { + return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS default verify paths")); + } + return {}; +} + +auto load_dtls_identity(SSL_CTX *context, ligo::security_config const &config) -> std::expected { + if (config.certificate_chain_file.empty() && config.private_key_file.empty()) { + return {}; + } + if (SSL_CTX_use_certificate_chain_file(context, config.certificate_chain_file.c_str()) != 1) { + return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS certificate chain")); + } + if (SSL_CTX_use_PrivateKey_file(context, config.private_key_file.c_str(), SSL_FILETYPE_PEM) != 1) { + return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS private key")); + } + if (SSL_CTX_check_private_key(context) != 1) { + return ligo::detail::fail_void(dtls_failure(ligo::detail::error_invalid_argument, "UDP DTLS private key does not match certificate")); + } + return {}; +} + +auto make_dtls_state(ligo::security_config const &config, std::bool_t server) -> std::expected, modulus::error> { + if (config.mode != ligo::security_mode::DTLS) { + return modulus::success(std::unique_ptr{}); + } + auto valid = validate_security(config, server); + if (!valid) { + return ligo::detail::fail>(valid.error()); + } + auto context = std::unique_ptr{SSL_CTX_new(server ? DTLS_server_method() : DTLS_client_method())}; + if (!context) { + return ligo::detail::fail>(ligo::detail::error_unavailable, "failed to create UDP DTLS context"); + } + SSL_CTX_set_options(context.get(), SSL_OP_NO_COMPRESSION | SSL_OP_NO_QUERY_MTU); + if (SSL_CTX_set_min_proto_version(context.get(), DTLS1_2_VERSION) != 1) { + return ligo::detail::fail>(dtls_failure("failed to configure UDP DTLS minimum protocol")); + } + auto const verify_peer = config.verify == ligo::verify_mode::PEER || (!server && config.verify == ligo::verify_mode::DEFAULTED); + auto const verify = verify_peer ? SSL_VERIFY_PEER | (server ? SSL_VERIFY_FAIL_IF_NO_PEER_CERT : 0) : SSL_VERIFY_NONE; + SSL_CTX_set_verify(context.get(), verify, nullptr); + auto verify_paths = load_dtls_verify_paths(context.get(), config); + if (!verify_paths) { + return ligo::detail::fail>(verify_paths.error()); + } + auto identity = load_dtls_identity(context.get(), config); + if (!identity) { + return ligo::detail::fail>(identity.error()); + } + auto ssl = std::unique_ptr{SSL_new(context.get())}; + if (!ssl) { + return ligo::detail::fail>(ligo::detail::error_unavailable, "failed to create UDP DTLS session"); + } + if (!server && verify_peer && !config.peer_name.empty()) { + if (X509_VERIFY_PARAM_set1_host(SSL_get0_param(ssl.get()), config.peer_name.c_str(), 0) != 1) { + return ligo::detail::fail>(dtls_failure("failed to configure UDP DTLS peer identity")); + } + } + SSL_set_mtu(ssl.get(), 1200); + auto *ssl_bio = static_cast(nullptr); + auto *network_bio = static_cast(nullptr); + if (BIO_new_bio_dgram_pair(&ssl_bio, 0, &network_bio, 0) != 1 || ssl_bio == nullptr || network_bio == nullptr) { + BIO_free(ssl_bio); + BIO_free(network_bio); + return ligo::detail::fail>(dtls_failure("failed to create UDP DTLS BIO pair")); + } + BIO_up_ref(ssl_bio); + SSL_set_bio(ssl.get(), ssl_bio, ssl_bio); + if (server) { + SSL_set_accept_state(ssl.get()); + } else { + SSL_set_connect_state(ssl.get()); + } + return modulus::success(std::make_unique(dtls_state{ + .context = std::move(context), + .ssl = std::move(ssl), + .network_bio = std::unique_ptr{network_bio}, + .server = server, + })); +} + +auto drain_dtls_output(dtls_state &state) -> std::expected, modulus::error> { + auto output = std::vector{}; + auto *bio = state.network_bio.get(); + while (BIO_ctrl_pending(bio) > 0) { + auto const pending = static_cast(BIO_ctrl_pending(bio)); + auto datagram = std::string(std::min(pending, max_datagram_size), '\0'); + auto const read = BIO_read(bio, datagram.data(), static_cast(datagram.size())); + if (read <= 0) { + return ligo::detail::fail>(dtls_failure("failed to read UDP DTLS output")); + } + datagram.resize(static_cast(read)); + output.push_back(std::move(datagram)); + } + return modulus::success(std::move(output)); +} + +auto append_dtls_output(std::vector &target, dtls_state &state) -> std::expected { + auto drained = drain_dtls_output(state); + if (!drained) { + return ligo::detail::fail_void(drained.error()); + } + for (auto &datagram : *drained) { + target.push_back(std::move(datagram)); + } + return {}; +} + +auto run_dtls_handshake(dtls_state &state, std::vector &output) -> std::expected { + if (state.handshake_complete) { + return {}; + } + state.handshake_started = true; + auto const result = SSL_do_handshake(state.ssl.get()); + if (result == 1) { + state.handshake_complete = true; + return append_dtls_output(output, state); + } + auto const error = SSL_get_error(state.ssl.get(), result); + if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE) { + return append_dtls_output(output, state); + } + return ligo::detail::fail_void(dtls_failure("UDP DTLS handshake failed")); +} + +auto protect_dtls_payload(dtls_state *state, std::string_view plaintext) -> std::expected, modulus::error> { + if (!state) { + return modulus::success(std::vector{ + std::string{plaintext}, + }); + } + if (plaintext.size() > max_dtls_plaintext_size) { + return ligo::detail::fail>(ligo::detail::error_invalid_argument, "UDP DTLS payload is too large"); + } + auto output = std::vector{}; + auto handshake = run_dtls_handshake(*state, output); + if (!handshake) { + return ligo::detail::fail>(handshake.error()); + } + if (!state->handshake_complete) { + return modulus::success(std::move(output)); + } + auto const written = SSL_write(state->ssl.get(), plaintext.data(), static_cast(plaintext.size())); + if (written <= 0) { + auto const error = SSL_get_error(state->ssl.get(), written); + if (error != SSL_ERROR_WANT_READ && error != SSL_ERROR_WANT_WRITE) { + return ligo::detail::fail>(dtls_failure("UDP DTLS write failed")); + } + } + auto drained = append_dtls_output(output, *state); + if (!drained) { + return ligo::detail::fail>(drained.error()); + } + return modulus::success(std::move(output)); +} + +auto process_dtls_input(dtls_state *state, std::string_view datagram) -> std::expected { + auto processed = dtls_processed{}; + if (!state) { + processed.plaintext.push_back(std::string{datagram}); + return modulus::success(std::move(processed)); + } + auto const written = BIO_write(state->network_bio.get(), datagram.data(), static_cast(datagram.size())); + if (written <= 0) { + return ligo::detail::fail(dtls_failure("failed to write UDP DTLS input")); + } + auto handshake = run_dtls_handshake(*state, processed.wire); + if (!handshake) { + return ligo::detail::fail(handshake.error()); + } + if (!state->handshake_complete) { + return modulus::success(std::move(processed)); + } + for (;;) { + auto plaintext = std::string(max_dtls_plaintext_size, '\0'); + auto const read = SSL_read(state->ssl.get(), plaintext.data(), static_cast(plaintext.size())); + if (read > 0) { + plaintext.resize(static_cast(read)); + processed.plaintext.push_back(std::move(plaintext)); + continue; + } + auto const error = SSL_get_error(state->ssl.get(), read); + if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE || error == SSL_ERROR_ZERO_RETURN) { + break; + } + return ligo::detail::fail(dtls_failure("UDP DTLS read failed")); + } + auto drained = append_dtls_output(processed.wire, *state); + if (!drained) { + return ligo::detail::fail(drained.error()); + } + return modulus::success(std::move(processed)); +} + +auto ssl_ctx_deleter::operator()(SSL_CTX *context) const noexcept -> void { + SSL_CTX_free(context); +} + +auto ssl_deleter::operator()(SSL *ssl) const noexcept -> void { + SSL_free(ssl); +} + +auto bio_deleter::operator()(BIO *bio) const noexcept -> void { + BIO_free(bio); +} + +} diff --git a/src/detail/dtls.hpp b/src/detail/dtls.hpp new file mode 100644 index 0000000..a47e2a7 --- /dev/null +++ b/src/detail/dtls.hpp @@ -0,0 +1,46 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +namespace ligo::detail { + +struct ssl_ctx_deleter { + auto operator()(SSL_CTX *context) const noexcept -> void; +}; + +struct ssl_deleter { + auto operator()(SSL *ssl) const noexcept -> void; +}; + +struct bio_deleter { + auto operator()(BIO *bio) const noexcept -> void; +}; + +struct dtls_state { + std::unique_ptr context{}; + std::unique_ptr ssl{}; + std::unique_ptr network_bio{}; + std::bool_t server{}; + std::bool_t handshake_started{}; + std::bool_t handshake_complete{}; +}; + +struct dtls_processed { + std::vector wire{}; + std::vector plaintext{}; +}; + +auto make_dtls_state(ligo::security_config const &config, std::bool_t server) -> std::expected, modulus::error>; +auto protect_dtls_payload(dtls_state *state, std::string_view plaintext) -> std::expected, modulus::error>; +auto process_dtls_input(dtls_state *state, std::string_view datagram) -> std::expected; + +} diff --git a/src/detail/udp_protocol.cpp b/src/detail/udp_protocol.cpp new file mode 100644 index 0000000..b274dd4 --- /dev/null +++ b/src/detail/udp_protocol.cpp @@ -0,0 +1,64 @@ +#include "udp_protocol.hpp" + +#include +#include +#include +#include +#include + + +namespace ligo::detail { + +auto uint_member(boost::json::object const &object, std::string_view name) -> std::optional { + auto found = object.find(name); + if (found == object.end() || !found->value().is_number()) { + return std::nullopt; + } + if (found->value().is_int64() && found->value().as_int64() < 0) { + return std::nullopt; + } + if (found->value().is_double()) { + return std::nullopt; + } + return found->value().is_uint64() ? found->value().as_uint64() : static_cast(found->value().as_int64()); +} + +auto string_member(boost::json::object const &object, std::string_view name) -> std::string { + auto found = object.find(name); + if (found == object.end() || !found->value().is_string()) { + return {}; + } + return std::string{found->value().as_string()}; +} + +auto make_data(std::uint64_t session_id, std::uint64_t sequence, boost::json::value payload) -> std::string { + auto envelope = boost::json::object{}; + envelope["kind"] = "data"; + envelope["session"] = session_id; + envelope["sequence"] = sequence; + envelope["payload"] = std::move(payload); + return boost::json::serialize(envelope); +} + +auto make_ack(std::uint64_t session_id, std::uint64_t sequence) -> std::string { + auto envelope = boost::json::object{}; + envelope["kind"] = "ack"; + envelope["session"] = session_id; + envelope["sequence"] = sequence; + return boost::json::serialize(envelope); +} + +auto validate_config(ligo::udp_reliability_config const &config) -> std::expected { + if (config.resend_interval <= std::chrono::milliseconds{0}) { + return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP resend_interval must be positive")); + } + if (config.max_in_flight == 0) { + return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP max_in_flight must be positive")); + } + if (config.session_timeout <= std::chrono::milliseconds{0}) { + return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP session_timeout must be positive")); + } + return {}; +} + +} diff --git a/src/detail/udp_protocol.hpp b/src/detail/udp_protocol.hpp new file mode 100644 index 0000000..2f71c95 --- /dev/null +++ b/src/detail/udp_protocol.hpp @@ -0,0 +1,22 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +namespace ligo::detail { + +auto uint_member(boost::json::object const &object, std::string_view name) -> std::optional; +auto string_member(boost::json::object const &object, std::string_view name) -> std::string; +auto make_data(std::uint64_t session_id, std::uint64_t sequence, boost::json::value payload) -> std::string; +auto make_ack(std::uint64_t session_id, std::uint64_t sequence) -> std::string; +auto validate_config(ligo::udp_reliability_config const &config) -> std::expected; + +} diff --git a/src/udp.cpp b/src/udp.cpp index 9c40950..51f9b7e 100755 --- a/src/udp.cpp +++ b/src/udp.cpp @@ -1,3 +1,5 @@ +#include "detail/dtls.hpp" +#include "detail/udp_protocol.hpp" #include #include @@ -51,41 +53,15 @@ namespace { -struct ssl_ctx_deleter { - auto operator()(SSL_CTX *context) const noexcept -> void; -}; - -struct ssl_deleter { - auto operator()(SSL *ssl) const noexcept -> void; -}; - -struct bio_deleter { - auto operator()(BIO *bio) const noexcept -> void; -}; - -struct dtls_state { - std::unique_ptr context{}; - std::unique_ptr ssl{}; - std::unique_ptr network_bio{}; - std::bool_t server{}; - std::bool_t handshake_started{}; - std::bool_t handshake_complete{}; -}; - -struct dtls_processed { - std::vector wire{}; - std::vector plaintext{}; -}; - struct reliable_channel_state { - reliable_channel_state(std::unique_ptr socket, boost::asio::ip::udp::endpoint local_endpoint, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::unique_ptr dtls); + reliable_channel_state(std::unique_ptr socket, boost::asio::ip::udp::endpoint local_endpoint, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::unique_ptr dtls); std::unique_ptr socket{}; boost::asio::strand strand{boost::asio::make_strand(socket->get_executor())}; boost::asio::ip::udp::endpoint local_endpoint{}; boost::asio::ip::udp::endpoint peer{}; ligo::udp_reliability_config config{}; ligo::security_config security{}; - std::unique_ptr dtls{}; + std::unique_ptr dtls{}; std::uint64_t session_id{1}; std::uint64_t next_send_sequence{1}; std::uint64_t next_receive_sequence{1}; @@ -107,300 +83,11 @@ struct reliable_channel_state { }; constexpr auto max_datagram_size = std::size_t{65507}; -constexpr auto max_dtls_plaintext_size = std::size_t{12000}; auto same_peer(boost::asio::ip::udp::endpoint const &left, boost::asio::ip::udp::endpoint const &right) -> std::bool_t { return left.address() == right.address() && left.port() == right.port(); } -auto uint_member(boost::json::object const &object, std::string_view name) -> std::optional { - auto found = object.find(name); - if (found == object.end() || !found->value().is_number()) { - return std::nullopt; - } - if (found->value().is_int64() && found->value().as_int64() < 0) { - return std::nullopt; - } - if (found->value().is_double()) { - return std::nullopt; - } - return found->value().is_uint64() ? found->value().as_uint64() : static_cast(found->value().as_int64()); -} - -auto string_member(boost::json::object const &object, std::string_view name) -> std::string { - auto found = object.find(name); - if (found == object.end() || !found->value().is_string()) { - return {}; - } - return std::string{found->value().as_string()}; -} - -auto make_data(std::uint64_t session_id, std::uint64_t sequence, boost::json::value payload) -> std::string { - auto envelope = boost::json::object{}; - envelope["kind"] = "data"; - envelope["session"] = session_id; - envelope["sequence"] = sequence; - envelope["payload"] = std::move(payload); - return boost::json::serialize(envelope); -} - -auto make_ack(std::uint64_t session_id, std::uint64_t sequence) -> std::string { - auto envelope = boost::json::object{}; - envelope["kind"] = "ack"; - envelope["session"] = session_id; - envelope["sequence"] = sequence; - return boost::json::serialize(envelope); -} - -auto validate_config(ligo::udp_reliability_config const &config) -> std::expected { - if (config.resend_interval <= std::chrono::milliseconds{0}) { - return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP resend_interval must be positive")); - } - if (config.max_in_flight == 0) { - return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP max_in_flight must be positive")); - } - if (config.session_timeout <= std::chrono::milliseconds{0}) { - return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP session_timeout must be positive")); - } - return {}; -} - -auto dtls_failure(std::int32_t code, std::string message) -> modulus::error { - auto const error = ERR_get_error(); - if (error != 0) { - message += ": "; - message += ERR_error_string(error, nullptr); - } - return ligo::detail::make_error(code, std::move(message)); -} - -auto dtls_failure(std::string message) -> modulus::error { - return dtls_failure(ligo::detail::error_unavailable, std::move(message)); -} - -auto validate_security(ligo::security_config const &config, std::bool_t server) -> std::expected { - if (config.mode == ligo::security_mode::TLS) { - return ligo::detail::fail_void(ligo::unsupported_security("UDP", config.mode)); - } - if (config.mode != ligo::security_mode::DTLS) { - return {}; - } - if (config.tls_context) { - return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS does not accept a TLS context")); - } - if (server && (config.certificate_chain_file.empty() || config.private_key_file.empty())) { - return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS server requires certificate chain/private key files")); - } - if (!config.certificate_chain_file.empty() != !config.private_key_file.empty()) { - return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS certificate chain and private key must both be configured")); - } - return {}; -} - -auto load_dtls_verify_paths(SSL_CTX *context, ligo::security_config const &config) -> std::expected { - if (config.verify == ligo::verify_mode::NONE) { - return {}; - } - if (!config.ca_file.empty()) { - if (SSL_CTX_load_verify_locations(context, config.ca_file.c_str(), nullptr) != 1) { - return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS CA file")); - } - return {}; - } - if (SSL_CTX_set_default_verify_paths(context) != 1) { - return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS default verify paths")); - } - return {}; -} - -auto load_dtls_identity(SSL_CTX *context, ligo::security_config const &config) -> std::expected { - if (config.certificate_chain_file.empty() && config.private_key_file.empty()) { - return {}; - } - if (SSL_CTX_use_certificate_chain_file(context, config.certificate_chain_file.c_str()) != 1) { - return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS certificate chain")); - } - if (SSL_CTX_use_PrivateKey_file(context, config.private_key_file.c_str(), SSL_FILETYPE_PEM) != 1) { - return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS private key")); - } - if (SSL_CTX_check_private_key(context) != 1) { - return ligo::detail::fail_void(dtls_failure(ligo::detail::error_invalid_argument, "UDP DTLS private key does not match certificate")); - } - return {}; -} - -auto make_dtls_state(ligo::security_config const &config, std::bool_t server) -> std::expected, modulus::error> { - if (config.mode != ligo::security_mode::DTLS) { - return modulus::success(std::unique_ptr{}); - } - auto valid = validate_security(config, server); - if (!valid) { - return ligo::detail::fail>(valid.error()); - } - auto context = std::unique_ptr{SSL_CTX_new(server ? DTLS_server_method() : DTLS_client_method())}; - if (!context) { - return ligo::detail::fail>(ligo::detail::error_unavailable, "failed to create UDP DTLS context"); - } - SSL_CTX_set_options(context.get(), SSL_OP_NO_COMPRESSION | SSL_OP_NO_QUERY_MTU); - if (SSL_CTX_set_min_proto_version(context.get(), DTLS1_2_VERSION) != 1) { - return ligo::detail::fail>(dtls_failure("failed to configure UDP DTLS minimum protocol")); - } - auto const verify_peer = config.verify == ligo::verify_mode::PEER || (!server && config.verify == ligo::verify_mode::DEFAULTED); - auto const verify = verify_peer ? SSL_VERIFY_PEER | (server ? SSL_VERIFY_FAIL_IF_NO_PEER_CERT : 0) : SSL_VERIFY_NONE; - SSL_CTX_set_verify(context.get(), verify, nullptr); - auto verify_paths = load_dtls_verify_paths(context.get(), config); - if (!verify_paths) { - return ligo::detail::fail>(verify_paths.error()); - } - auto identity = load_dtls_identity(context.get(), config); - if (!identity) { - return ligo::detail::fail>(identity.error()); - } - auto ssl = std::unique_ptr{SSL_new(context.get())}; - if (!ssl) { - return ligo::detail::fail>(ligo::detail::error_unavailable, "failed to create UDP DTLS session"); - } - if (!server && verify_peer && !config.peer_name.empty()) { - if (X509_VERIFY_PARAM_set1_host(SSL_get0_param(ssl.get()), config.peer_name.c_str(), 0) != 1) { - return ligo::detail::fail>(dtls_failure("failed to configure UDP DTLS peer identity")); - } - } - SSL_set_mtu(ssl.get(), 1200); - auto *ssl_bio = static_cast(nullptr); - auto *network_bio = static_cast(nullptr); - if (BIO_new_bio_dgram_pair(&ssl_bio, 0, &network_bio, 0) != 1 || ssl_bio == nullptr || network_bio == nullptr) { - BIO_free(ssl_bio); - BIO_free(network_bio); - return ligo::detail::fail>(dtls_failure("failed to create UDP DTLS BIO pair")); - } - BIO_up_ref(ssl_bio); - SSL_set_bio(ssl.get(), ssl_bio, ssl_bio); - if (server) { - SSL_set_accept_state(ssl.get()); - } else { - SSL_set_connect_state(ssl.get()); - } - return modulus::success(std::make_unique(dtls_state{ - .context = std::move(context), - .ssl = std::move(ssl), - .network_bio = std::unique_ptr{network_bio}, - .server = server, - })); -} - -auto drain_dtls_output(dtls_state &state) -> std::expected, modulus::error> { - auto output = std::vector{}; - auto *bio = state.network_bio.get(); - while (BIO_ctrl_pending(bio) > 0) { - auto const pending = static_cast(BIO_ctrl_pending(bio)); - auto datagram = std::string(std::min(pending, max_datagram_size), '\0'); - auto const read = BIO_read(bio, datagram.data(), static_cast(datagram.size())); - if (read <= 0) { - return ligo::detail::fail>(dtls_failure("failed to read UDP DTLS output")); - } - datagram.resize(static_cast(read)); - output.push_back(std::move(datagram)); - } - return modulus::success(std::move(output)); -} - -auto append_dtls_output(std::vector &target, dtls_state &state) -> std::expected { - auto drained = drain_dtls_output(state); - if (!drained) { - return ligo::detail::fail_void(drained.error()); - } - for (auto &datagram : *drained) { - target.push_back(std::move(datagram)); - } - return {}; -} - -auto run_dtls_handshake(dtls_state &state, std::vector &output) -> std::expected { - if (state.handshake_complete) { - return {}; - } - state.handshake_started = true; - auto const result = SSL_do_handshake(state.ssl.get()); - if (result == 1) { - state.handshake_complete = true; - return append_dtls_output(output, state); - } - auto const error = SSL_get_error(state.ssl.get(), result); - if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE) { - return append_dtls_output(output, state); - } - return ligo::detail::fail_void(dtls_failure("UDP DTLS handshake failed")); -} - -auto protect_dtls_payload(reliable_channel_state &state, std::string_view plaintext) -> std::expected, modulus::error> { - if (!state.dtls) { - return modulus::success(std::vector{ - std::string{plaintext}, - }); - } - if (plaintext.size() > max_dtls_plaintext_size) { - return ligo::detail::fail>(ligo::detail::error_invalid_argument, "UDP DTLS payload is too large"); - } - auto output = std::vector{}; - auto handshake = run_dtls_handshake(*state.dtls, output); - if (!handshake) { - return ligo::detail::fail>(handshake.error()); - } - if (!state.dtls->handshake_complete) { - return modulus::success(std::move(output)); - } - auto const written = SSL_write(state.dtls->ssl.get(), plaintext.data(), static_cast(plaintext.size())); - if (written <= 0) { - auto const error = SSL_get_error(state.dtls->ssl.get(), written); - if (error != SSL_ERROR_WANT_READ && error != SSL_ERROR_WANT_WRITE) { - return ligo::detail::fail>(dtls_failure("UDP DTLS write failed")); - } - } - auto drained = append_dtls_output(output, *state.dtls); - if (!drained) { - return ligo::detail::fail>(drained.error()); - } - return modulus::success(std::move(output)); -} - -auto process_dtls_input(reliable_channel_state &state, std::string_view datagram) -> std::expected { - auto processed = dtls_processed{}; - if (!state.dtls) { - processed.plaintext.push_back(std::string{datagram}); - return modulus::success(std::move(processed)); - } - auto const written = BIO_write(state.dtls->network_bio.get(), datagram.data(), static_cast(datagram.size())); - if (written <= 0) { - return ligo::detail::fail(dtls_failure("failed to write UDP DTLS input")); - } - auto handshake = run_dtls_handshake(*state.dtls, processed.wire); - if (!handshake) { - return ligo::detail::fail(handshake.error()); - } - if (!state.dtls->handshake_complete) { - return modulus::success(std::move(processed)); - } - for (;;) { - auto plaintext = std::string(max_dtls_plaintext_size, '\0'); - auto const read = SSL_read(state.dtls->ssl.get(), plaintext.data(), static_cast(plaintext.size())); - if (read > 0) { - plaintext.resize(static_cast(read)); - processed.plaintext.push_back(std::move(plaintext)); - continue; - } - auto const error = SSL_get_error(state.dtls->ssl.get(), read); - if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE || error == SSL_ERROR_ZERO_RETURN) { - break; - } - return ligo::detail::fail(dtls_failure("UDP DTLS read failed")); - } - auto drained = append_dtls_output(processed.wire, *state.dtls); - if (!drained) { - return ligo::detail::fail(drained.error()); - } - return modulus::success(std::move(processed)); -} - auto channel_state(ligo::detail::state_handle const &handle) -> std::shared_ptr { return handle.as(); } @@ -432,8 +119,8 @@ auto send_wire_datagrams(std::shared_ptr const &state, s } auto send_ack(std::shared_ptr const &state, std::uint64_t sequence) -> boost::asio::awaitable> { - auto ack_body = make_ack(state->session_id, sequence); - auto datagrams = protect_dtls_payload(*state, ack_body); + auto ack_body = ligo::detail::make_ack(state->session_id, sequence); + auto datagrams = ligo::detail::protect_dtls_payload(state->dtls.get(), ack_body); if (!datagrams) { co_return ligo::detail::fail_void(datagrams.error()); } @@ -487,10 +174,10 @@ auto receive_loop(std::shared_ptr state) -> boost::asio: if (size > max_datagram_size || !same_peer(state->receive_sender, state->peer)) { continue; } - auto processed = process_dtls_input(*state, std::string_view{ - state->receive_buffer.data(), - size, - }); + auto processed = ligo::detail::process_dtls_input(state->dtls.get(), std::string_view{ + state->receive_buffer.data(), + size, + }); if (!processed) { state->receive_error = processed.error(); notify_send_waiters(state); @@ -511,12 +198,12 @@ auto receive_loop(std::shared_ptr state) -> boost::asio: continue; } auto const &object = envelope.as_object(); - auto session = uint_member(object, "session"); - auto sequence = uint_member(object, "sequence"); + auto session = ligo::detail::uint_member(object, "session"); + auto sequence = ligo::detail::uint_member(object, "sequence"); if (!session.has_value() || *session != state->session_id || !sequence.has_value()) { continue; } - auto const kind = string_member(object, "kind"); + auto const kind = ligo::detail::string_member(object, "kind"); if (kind == "ack") { if (state->active_send_sequences.contains(*sequence)) { state->acked.insert(*sequence); @@ -617,11 +304,11 @@ auto udp_endpoint_from_parts(ligo::detail::endpoint_parts const &parts) -> std:: } auto open_state(boost::asio::any_io_executor executor, boost::asio::ip::udp::endpoint local, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::bool_t server) -> std::expected, modulus::error> { - auto valid = validate_config(config); + auto valid = ligo::detail::validate_config(config); if (!valid) { return ligo::detail::fail>(valid.error()); } - auto dtls = make_dtls_state(security, server); + auto dtls = ligo::detail::make_dtls_state(security, server); if (!dtls) { return ligo::detail::fail>(dtls.error()); } @@ -646,11 +333,11 @@ auto open_state(boost::asio::any_io_executor executor, boost::asio::ip::udp::end } auto adopt_state(std::unique_ptr socket, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::bool_t server) -> std::expected, modulus::error> { - auto valid = validate_config(config); + auto valid = ligo::detail::validate_config(config); if (!valid) { return ligo::detail::fail>(valid.error()); } - auto dtls = make_dtls_state(security, server); + auto dtls = ligo::detail::make_dtls_state(security, server); if (!dtls) { return ligo::detail::fail>(dtls.error()); } @@ -673,7 +360,7 @@ auto send_payload_on_strand(std::shared_ptr state, boost co_return ligo::detail::fail_void(ligo::detail::unavailable("too many UDP messages in flight")); } auto const sequence = state->next_send_sequence; - auto body = make_data(state->session_id, sequence, std::move(payload)); + auto body = ligo::detail::make_data(state->session_id, sequence, std::move(payload)); if (body.size() > max_datagram_size) { co_return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP payload is too large")); } @@ -691,7 +378,7 @@ auto send_payload_on_strand(std::shared_ptr state, boost finish_send(state, sequence); co_return ligo::detail::fail_void(ligo::detail::unavailable("UDP channel closed")); } - auto datagrams = protect_dtls_payload(*state, body); + auto datagrams = ligo::detail::protect_dtls_payload(state->dtls.get(), body); if (!datagrams) { finish_send(state, sequence); co_return ligo::detail::fail_void(datagrams.error()); @@ -899,19 +586,7 @@ auto endpoint_from_config(ligo::endpoint_config const &config) -> std::expected< }); } -auto ssl_ctx_deleter::operator()(SSL_CTX *context) const noexcept -> void { - SSL_CTX_free(context); -} - -auto ssl_deleter::operator()(SSL *ssl) const noexcept -> void { - SSL_free(ssl); -} - -auto bio_deleter::operator()(BIO *bio) const noexcept -> void { - BIO_free(bio); -} - -reliable_channel_state::reliable_channel_state(std::unique_ptr socket, boost::asio::ip::udp::endpoint local_endpoint, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::unique_ptr dtls) +reliable_channel_state::reliable_channel_state(std::unique_ptr socket, boost::asio::ip::udp::endpoint local_endpoint, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::unique_ptr dtls) : socket{std::move(socket)}, local_endpoint{std::move(local_endpoint)}, peer{std::move(peer)}, config{config}, security{std::move(security)}, dtls{std::move(dtls)}, session_id{session_id} { notify_timer.expires_at(std::chrono::steady_clock::time_point::max()); close_timer.expires_at(std::chrono::steady_clock::time_point::max()); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index a767248..239e769 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -1,5 +1,6 @@ add_executable(ligo_test test.cpp + lifecycle.cpp ) find_package(Threads REQUIRED) target_include_directories(ligo_test PRIVATE diff --git a/tests/lifecycle.cpp b/tests/lifecycle.cpp new file mode 100644 index 0000000..a6377f1 --- /dev/null +++ b/tests/lifecycle.cpp @@ -0,0 +1,75 @@ +#include +#include +#include +#include +#include +#include +#include +#include + + +BOOST_AUTO_TEST_CASE(session_drain_handles_notifications_before_and_during_registration) { + for (int iteration = 0; iteration < 64; ++iteration) { + auto io = boost::asio::io_context{}; + auto registry = ligo::detail::session_registry{}; + auto session = std::make_shared(1); + registry.add(session); + auto result = boost::asio::co_spawn(io, registry.wait_empty(std::chrono::seconds{2}), boost::asio::use_future); + auto work = boost::asio::make_work_guard(io); + auto first = std::thread{[&] { io.run(); }}; + auto second = std::thread{[&] { io.run(); }}; + registry.remove(session); + auto ready = result.wait_for(std::chrono::milliseconds{500}) == std::future_status::ready; + work.reset(); + io.stop(); + first.join(); + second.join(); + BOOST_REQUIRE(ready); + BOOST_CHECK(result.get()); + } +} + +BOOST_AUTO_TEST_CASE(session_drain_keeps_absolute_deadline_under_repeated_notifications) { + auto io = boost::asio::io_context{}; + auto registry = ligo::detail::session_registry{}; + auto session = std::make_shared(1); + registry.add(session); + auto result = boost::asio::co_spawn(io, registry.wait_empty(std::chrono::milliseconds{20}), boost::asio::use_future); + auto work = boost::asio::make_work_guard(io); + auto first = std::thread{[&] { io.run(); }}; + auto second = std::thread{[&] { io.run(); }}; + for (int i = 0; i < 200 && result.wait_for(std::chrono::milliseconds{1}) != std::future_status::ready; ++i) { + registry.notify_waiters(); + } + auto ready = result.wait_for(std::chrono::milliseconds{0}) == std::future_status::ready; + work.reset(); + io.stop(); + first.join(); + second.join(); + BOOST_REQUIRE(ready); + auto value = result.get(); + BOOST_REQUIRE(!value); + BOOST_CHECK(value.error().code() == ligo::detail::error_timeout); +} + +BOOST_AUTO_TEST_CASE(session_drain_observes_expiry_and_removal_before_wait_is_armed) { + auto io = boost::asio::io_context{}; + auto registry = ligo::detail::session_registry{}; + auto session = std::make_shared(1); + registry.add(session); + auto timer = std::make_shared(boost::asio::make_strand(io)); + registry.waiters.push_back(timer); + // A notification is queued before registration. It must neither extend an + // elapsed deadline nor prevent observing a session that already finished. + registry.notify_waiters(); + auto expired = boost::asio::co_spawn(timer->get_executor(), registry.wait_registered(timer, std::chrono::steady_clock::now()), boost::asio::use_future); + io.run(); + auto result = expired.get(); + BOOST_REQUIRE(!result); + BOOST_CHECK(result.error().code() == ligo::detail::error_timeout); + io.restart(); + registry.remove(session); + auto drained = boost::asio::co_spawn(timer->get_executor(), registry.wait_registered(timer, std::chrono::steady_clock::now() + std::chrono::hours{1}), boost::asio::use_future); + io.run(); + BOOST_CHECK(drained.get()); +}