Merge pull request #11134

5066fd0 wallet2: abort in-flight daemon request on shutdown (woodser)

ACKs: amin404ala*, thomasbuilds, selsta
This commit is contained in:
tobtoht
2026-08-26 07:33:16 +00:00
9 changed files with 360 additions and 27 deletions
@@ -68,6 +68,7 @@ namespace http
virtual void set_auto_connect(bool auto_connect) = 0;
virtual bool connect(std::chrono::milliseconds timeout) = 0;
virtual bool disconnect() = 0;
virtual bool shutdown() { return disconnect(); } // tear down: the client need not be usable afterwards; overrides should interrupt blocked calls and be callable from another thread (the default plain disconnect() gives neither guarantee)
virtual bool is_connected(bool *ssl = NULL) = 0;
virtual bool invoke(const boost::string_ref uri, const boost::string_ref method, const boost::string_ref body, std::chrono::milliseconds timeout, const http_response_info** ppresponse_info = NULL, const fields_list& additional_params = fields_list()) = 0;
virtual bool invoke_get(const boost::string_ref uri, std::chrono::milliseconds timeout, const std::string& body = std::string(), const http_response_info** ppresponse_info = NULL, const fields_list& additional_params = fields_list()) = 0;
+5
View File
@@ -154,6 +154,11 @@ namespace net_utils
return m_net_client.disconnect();
}
//---------------------------------------------------------------------------
bool shutdown() override
{
return m_net_client.shutdown(); // no m_lock; interrupts an invoke blocked while holding it
}
//---------------------------------------------------------------------------
bool is_connected(bool *ssl = NULL) override
{
CRITICAL_REGION_LOCAL(m_lock);
+135 -24
View File
@@ -34,6 +34,7 @@
#include <boost/version.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/read.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/asio/steady_timer.hpp>
@@ -43,6 +44,13 @@
#include <boost/system/error_code.hpp>
#include <boost/utility/string_ref.hpp>
#include <functional>
#ifdef BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR
#include <boost/asio/posix/stream_descriptor.hpp>
#include <cerrno>
#include <fcntl.h>
#include <stdexcept>
#include <unistd.h>
#endif
#include "net/net_ssl.h"
#include "misc_log_ex.h"
@@ -101,11 +109,39 @@ namespace net_utils
m_ssl_options(epee::net_utils::ssl_support_t::e_ssl_support_autodetect),
m_connected(false),
m_deadline(m_io_service, std::chrono::steady_clock::time_point::max()),
m_shutdowned(false),
#ifdef BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR
m_wake_reader(m_io_service),
m_wake_writer(-1),
#endif
m_aborted(false),
m_bytes_sent(0),
m_bytes_received(0)
{
check_deadline();
#ifdef BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR
// self-pipe so shutdown() can wake a blocked run_one() with an async-signal-safe
// write(); without it aborts would wait out the op deadline, so refuse to construct
int fds[2];
if (0 != ::pipe(fds))
throw std::runtime_error("failed to create shutdown wake pipe");
bool flags_set = true;
for (const int fd : fds)
{
if (-1 == ::fcntl(fd, F_SETFL, O_NONBLOCK) || -1 == ::fcntl(fd, F_SETFD, FD_CLOEXEC))
flags_set = false;
}
boost::system::error_code ec;
if (flags_set)
m_wake_reader.assign(fds[0], ec);
if (!flags_set || ec)
{
::close(fds[0]);
::close(fds[1]);
throw std::runtime_error("failed to set up shutdown wake pipe");
}
m_wake_writer = fds[1];
wait_wake();
#endif
}
/*! The first/second parameters are host/port respectively. The third
@@ -128,8 +164,12 @@ namespace net_utils
~blocked_mode_client()
{
//profile_tools::local_coast lc("~blocked_mode_client()", 3);
try { shutdown(); }
try { disconnect(); }
catch(...) { /* ignore */ }
#ifdef BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR
if (m_wake_writer != -1)
::close(m_wake_writer); // read end is owned and closed by m_wake_reader
#endif
}
inline void set_ssl(ssl_options_t ssl_options)
@@ -144,10 +184,16 @@ namespace net_utils
inline
try_connect_result_t try_connect(const std::string& addr, const std::string& port, std::chrono::milliseconds timeout)
{
const auto deadline = std::chrono::steady_clock::now() + timeout;
m_deadline.expires_after(timeout);
boost::unique_future<boost::asio::ip::tcp::socket> connection = m_connector(addr, port, m_deadline);
for (;;)
{
// aborted or timed out: abandon the attempt; its handlers own their state
// and their socket dies with them, at completion or client destruction
if (shutdown_requested() || std::chrono::steady_clock::now() >= deadline)
return CONNECT_FAILURE;
m_io_service.restart();
m_io_service.run_one();
@@ -157,6 +203,12 @@ namespace net_utils
m_ssl_socket->next_layer() = connection.get();
m_deadline.cancel();
if (shutdown_requested())
{
boost::system::error_code ignored_ec;
m_ssl_socket->next_layer().close(ignored_ec);
return CONNECT_FAILURE; // aborted by shutdown() from another thread
}
if (m_ssl_socket->next_layer().is_open())
{
m_connected = true;
@@ -164,8 +216,13 @@ namespace net_utils
// SSL Options
if (m_ssl_options.support == epee::net_utils::ssl_support_t::e_ssl_support_enabled || m_ssl_options.support == epee::net_utils::ssl_support_t::e_ssl_support_autodetect)
{
if (!m_ssl_options.handshake(m_io_service, *m_ssl_socket, boost::asio::ssl::stream_base::client, {}, addr, timeout))
if (!m_ssl_options.handshake(m_io_service, *m_ssl_socket, boost::asio::ssl::stream_base::client, {}, addr, timeout, [this] { return shutdown_requested(); }))
{
if (shutdown_requested())
{
m_connected = false;
return CONNECT_FAILURE; // aborted: says nothing about the server's TLS support
}
if (m_ssl_options.support == epee::net_utils::ssl_support_t::e_ssl_support_autodetect)
{
boost::system::error_code ignored_ec;
@@ -182,6 +239,15 @@ namespace net_utils
}
}
}
// recheck after publishing m_connected: shutdown() between the check above
// and the publish must not leave a live connection behind
if (shutdown_requested())
{
m_connected = false;
boost::system::error_code ignored_ec;
m_ssl_socket->next_layer().close(ignored_ec);
return CONNECT_FAILURE;
}
return CONNECT_SUCCESS;
}else
{
@@ -194,6 +260,8 @@ namespace net_utils
inline
bool connect(const std::string& addr, const std::string& port, std::chrono::milliseconds timeout)
{
if (shutdown_requested())
return false; // torn down: the client never reconnects after shutdown()
m_connected = false;
try
{
@@ -268,7 +336,6 @@ namespace net_utils
inline
bool send(const boost::string_ref buff, std::chrono::milliseconds timeout)
{
try
{
m_deadline.expires_after(timeout);
@@ -287,12 +354,25 @@ namespace net_utils
async_write(buff.data(), buff.size(), ec);
// Block until the asynchronous operation has completed.
while (ec == boost::asio::error::would_block)
while (ec == boost::asio::error::would_block && !shutdown_requested())
{
m_io_service.restart();
m_io_service.run_one();
}
// aborted by shutdown(): the pending handler references this frame, so
// close the socket and pump until it completes before returning
if (ec == boost::asio::error::would_block)
{
boost::system::error_code ignored_ec;
m_ssl_socket->next_layer().close(ignored_ec);
while (ec == boost::asio::error::would_block)
{
m_io_service.restart();
m_io_service.run_one();
}
}
if (ec)
{
LOG_PRINT_L3("Problems at write: " << ec.message());
@@ -331,7 +411,6 @@ namespace net_utils
inline
bool recv(std::string& buff, std::chrono::milliseconds timeout)
{
try
{
// Set a deadline for the asynchronous operation. Since this function uses
@@ -362,12 +441,24 @@ namespace net_utils
async_read(&buff[0], max_size, boost::asio::transfer_at_least(1), hndlr);
// Block until the asynchronous operation has completed.
while (ec == boost::asio::error::would_block && !m_shutdowned)
while (ec == boost::asio::error::would_block && !shutdown_requested())
{
m_io_service.restart();
m_io_service.run_one();
}
// aborted by shutdown(): the pending handler references this frame, so
// close the socket and pump until it completes before returning
if (ec == boost::asio::error::would_block)
{
boost::system::error_code ignored_ec;
m_ssl_socket->next_layer().close(ignored_ec);
while (ec == boost::asio::error::would_block)
{
m_io_service.restart();
m_io_service.run_one();
}
}
if (ec)
{
@@ -417,21 +508,20 @@ namespace net_utils
bool shutdown()
{
m_deadline.cancel();
boost::system::error_code ec;
if(m_ssl_options)
shutdown_ssl();
m_ssl_socket->next_layer().cancel(ec);
if(ec)
MDEBUG("Problems at cancel: " << ec.message());
m_ssl_socket->next_layer().shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
if(ec)
MDEBUG("Problems at shutdown: " << ec.message());
m_ssl_socket->next_layer().close(ec);
if(ec)
MDEBUG("Problems at close: " << ec.message());
m_shutdowned = true;
m_connected = false;
// callable from any thread, on POSIX even a signal handler: only atomics and a self-pipe
// write(); a blocked call closes the socket on abort, an idle client's closes at destruction
m_aborted = true;
m_connected = false;
#ifdef BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR
// construction guarantees the pipe exists
const int saved_errno = errno;
const char wake = 0;
while (::write(m_wake_writer, &wake, 1) == -1 && errno == EINTR) {}
errno = saved_errno; // a full pipe already holds a pending wake
#else
// Windows console handlers run on their own thread: post() a wake
boost::asio::post(m_io_service, []{});
#endif
return true;
}
@@ -447,6 +537,20 @@ namespace net_utils
private:
bool shutdown_requested() const
{
return m_aborted.load();
}
#ifdef BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR
// perpetual pipe read so a wake byte written by shutdown() unblocks run_one()
void wait_wake()
{
m_wake_reader.async_read_some(boost::asio::buffer(m_wake_buf),
[this] (const boost::system::error_code& ec, std::size_t) { if (!ec) wait_wake(); });
}
#endif
void check_deadline()
{
// Check whether the deadline has passed. We compare the deadline against
@@ -516,9 +620,16 @@ namespace net_utils
std::shared_ptr<boost::asio::ssl::stream<boost::asio::ip::tcp::socket>> m_ssl_socket;
std::function<connect_func> m_connector;
ssl_options_t m_ssl_options;
bool m_connected;
std::atomic<bool> m_connected;
boost::asio::steady_timer m_deadline;
std::atomic<bool> m_shutdowned;
#ifdef BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR
// self-pipe: shutdown() write()s the pipe to wake run_one() signal-safely
boost::asio::posix::stream_descriptor m_wake_reader;
int m_wake_writer;
char m_wake_buf[8];
#endif
// sticky: set once by shutdown(), never cleared; the client is being torn down
std::atomic<bool> m_aborted;
std::atomic<uint64_t> m_bytes_sent;
std::atomic<uint64_t> m_bytes_received;
};
+5 -1
View File
@@ -30,6 +30,7 @@
#define _NET_SSL_H
#include <chrono>
#include <functional>
#include <stdint.h>
#include <string>
#include <vector>
@@ -135,6 +136,8 @@ namespace net_utils
situations where multiple hostnames are being handled by a server. If
`verification == system_ca` the client also does a rfc2818 check to
ensure that the server certificate is to the provided hostname.
\param abort_requested Polled while waiting; when it returns true the
handshake is cancelled and fails instead of running to `timeout`.
\return True if the SSL handshake completes with peer verification
settings. */
@@ -144,7 +147,8 @@ namespace net_utils
boost::asio::ssl::stream_base::handshake_type type,
boost::asio::const_buffer buffer = {},
const std::string& host = {},
std::chrono::milliseconds timeout = std::chrono::seconds(15)) const;
std::chrono::milliseconds timeout = std::chrono::seconds(15),
const std::function<bool()>& abort_requested = nullptr) const;
};
// https://security.stackexchange.com/questions/34780/checking-client-hello-for-https-classification
+29 -1
View File
@@ -441,7 +441,8 @@ bool ssl_options_t::handshake(
boost::asio::ssl::stream_base::handshake_type type,
boost::asio::const_buffer buffer,
const std::string& host,
std::chrono::milliseconds timeout) const
std::chrono::milliseconds timeout,
const std::function<bool()>& abort_requested) const
{
configure(socket, type, host);
@@ -495,6 +496,15 @@ bool ssl_options_t::handshake(
boost::asio::post(
strand,
[&]{
std::lock_guard<std::mutex> guard(state.lock);
if (state.cancel_handshake) {
// cancelled before the socket operation could start: finish here so
// the cancelling side cannot leave a handshake running unattended
state.wait_handshake = false;
state.result = boost::asio::error::operation_aborted;
state.condition.notify_all();
return;
}
socket.async_handshake(
type,
boost::asio::buffer(buffer),
@@ -503,10 +513,28 @@ bool ssl_options_t::handshake(
}
);
bool aborted = false;
while (!io_context.stopped())
{
io_context.poll_one();
std::lock_guard<std::mutex> guard(state.lock);
if (!aborted && abort_requested && abort_requested())
{
// abort from another thread: close the socket so the handshake cannot
// stall again, then keep pumping until both handlers have finished
aborted = true;
if (state.wait_handshake && !state.cancel_handshake)
{
state.cancel_handshake = true;
ec_t ec;
socket.next_layer().close(ec);
}
if (state.wait_timer && !state.cancel_timer)
{
state.cancel_timer = true;
deadline.cancel();
}
}
state.condition.wait_for(
state.lock,
std::chrono::milliseconds(30),
+2 -1
View File
@@ -797,7 +797,8 @@ private:
bool set_proxy(const std::string &address);
void stop() { m_run.store(false, std::memory_order_relaxed); m_message_store.stop(); }
void shutdown() { m_stopped.store(true, std::memory_order_relaxed); stop(); }
// teardown-only: a permanent stop that also aborts an in-flight daemon request
void shutdown() { m_stopped.store(true, std::memory_order_relaxed); stop(); m_http_client->shutdown(); }
i_wallet2_callback* callback() const { return m_callback; }
void callback(i_wallet2_callback* callback) { m_callback = callback; }
+1
View File
@@ -39,6 +39,7 @@ public:
bool connect(const std::string& addr, int port, std::chrono::milliseconds timeout, bool ssl = false, const std::string& bind_ip = "0.0.0.0") { return true; }
bool connect(const std::string& addr, const std::string& port, std::chrono::milliseconds timeout, bool ssl = false, const std::string& bind_ip = "0.0.0.0") { return true; }
bool disconnect() { return true; }
bool shutdown() { return true; }
bool send(const boost::string_ref buff, std::chrono::milliseconds timeout) { return true; }
bool send(const void* data, size_t sz) { return true; }
bool is_connected() { return true; }
+1
View File
@@ -95,6 +95,7 @@ public:
bool connect(const std::string&, int, std::chrono::milliseconds, bool = false, const std::string& = "0.0.0.0") { return true; }
bool connect(const std::string&, const std::string&, std::chrono::milliseconds, bool = false, const std::string& = "0.0.0.0") { return true; }
bool disconnect() { return true; }
bool shutdown() { return true; }
bool send(const boost::string_ref, std::chrono::milliseconds) { return true; }
bool send(const void*, size_t) { return true; }
bool recv(std::string& buff, std::chrono::milliseconds)
+181
View File
@@ -51,13 +51,20 @@
#include <gtest/gtest.h>
#include <map>
#include <memory>
#include <thread>
#include <type_traits>
#ifndef _WIN32
#include <atomic>
#include <csignal>
#include <pthread.h>
#endif
#include "crypto/crypto.h"
#include "net/dandelionpp.h"
#include "net/error.h"
#include "net/host.h"
#include "net/i2p_address.h"
#include "net/net_helper.h"
#include "net/net_utils_base.h"
#include "net/socks.h"
#include "net/socks_connect.h"
@@ -94,6 +101,180 @@ namespace
"zpv4fa3szgel7vf6jdjeugizdclq2vzkelscs2bhbgnlldzzggcen3ac.onion";
}
TEST(blocked_mode_client, shutdown_aborts_blocked_recv)
{
boost::asio::io_context server_io;
boost::asio::ip::tcp::acceptor acceptor{
server_io,
{boost::asio::ip::address_v4::loopback(), 0}
};
epee::net_utils::blocked_mode_client client;
client.set_ssl(epee::net_utils::ssl_options_t{
epee::net_utils::ssl_support_t::e_ssl_support_disabled
});
const std::string port = std::to_string(acceptor.local_endpoint().port());
ASSERT_TRUE(client.connect("127.0.0.1", port, std::chrono::seconds{5}));
boost::system::error_code error;
boost::asio::ip::tcp::socket peer{server_io};
acceptor.accept(peer, error);
ASSERT_FALSE(error);
// recv blocks on the silent peer until a single shutdown from another thread
// aborts it; the abort is sticky, so the shot is valid on every timing
std::thread stopper{[&client] {
std::this_thread::sleep_for(std::chrono::milliseconds{100});
EXPECT_TRUE(client.shutdown());
}};
std::string response;
const auto start = std::chrono::steady_clock::now();
EXPECT_FALSE(client.recv(response, std::chrono::seconds{30}));
EXPECT_LT(std::chrono::steady_clock::now() - start, std::chrono::seconds{10});
stopper.join();
EXPECT_FALSE(client.is_connected());
}
#ifndef _WIN32
namespace
{
std::atomic<epee::net_utils::blocked_mode_client*> g_signal_client{nullptr};
void shutdown_from_signal(int)
{
epee::net_utils::blocked_mode_client* const client = g_signal_client.load();
if (client)
client->shutdown();
}
}
TEST(blocked_mode_client, shutdown_aborts_blocked_recv_from_signal_handler)
{
boost::asio::io_context server_io;
boost::asio::ip::tcp::acceptor acceptor{
server_io,
{boost::asio::ip::address_v4::loopback(), 0}
};
epee::net_utils::blocked_mode_client client;
client.set_ssl(epee::net_utils::ssl_options_t{
epee::net_utils::ssl_support_t::e_ssl_support_disabled
});
const std::string port = std::to_string(acceptor.local_endpoint().port());
ASSERT_TRUE(client.connect("127.0.0.1", port, std::chrono::seconds{5}));
boost::system::error_code error;
boost::asio::ip::tcp::socket peer{server_io};
acceptor.accept(peer, error);
ASSERT_FALSE(error);
struct sigaction handler {}, previous {};
handler.sa_handler = shutdown_from_signal;
sigemptyset(&handler.sa_mask);
ASSERT_EQ(0, sigaction(SIGUSR1, &handler, &previous));
g_signal_client = &client;
// the signal lands on the thread blocked in recv, so the handler's shutdown()
// runs in signal context and may only use async-signal-safe calls
const pthread_t blocked = pthread_self();
std::thread stopper{[blocked] {
std::this_thread::sleep_for(std::chrono::milliseconds{100});
pthread_kill(blocked, SIGUSR1);
}};
std::string response;
const auto start = std::chrono::steady_clock::now();
EXPECT_FALSE(client.recv(response, std::chrono::seconds{30}));
EXPECT_LT(std::chrono::steady_clock::now() - start, std::chrono::seconds{10});
stopper.join();
EXPECT_FALSE(client.is_connected());
g_signal_client = nullptr;
sigaction(SIGUSR1, &previous, nullptr);
}
#endif
TEST(blocked_mode_client, shutdown_aborts_blocked_connect)
{
// a connector whose future never becomes ready simulates an unresponsive peer
const auto never_ready = std::make_shared<boost::promise<boost::asio::ip::tcp::socket>>();
epee::net_utils::blocked_mode_client client;
client.set_ssl(epee::net_utils::ssl_options_t{
epee::net_utils::ssl_support_t::e_ssl_support_disabled
});
client.set_connector([never_ready] (const std::string&, const std::string&, boost::asio::steady_timer&) {
return never_ready->get_future();
});
// a single shutdown aborts the blocked connect; sticky, so valid on every timing
std::thread stopper{[&client] {
std::this_thread::sleep_for(std::chrono::milliseconds{100});
EXPECT_TRUE(client.shutdown());
}};
const auto start = std::chrono::steady_clock::now();
EXPECT_FALSE(client.connect("127.0.0.1", "80", std::chrono::seconds{30}));
EXPECT_LT(std::chrono::steady_clock::now() - start, std::chrono::seconds{10});
stopper.join();
}
TEST(blocked_mode_client, shutdown_aborts_stalled_ssl_handshake)
{
// the acceptor completes TCP connects in the kernel backlog but never
// answers the TLS handshake, so the handshake stalls until aborted
boost::asio::io_context server_io;
boost::asio::ip::tcp::acceptor acceptor{
server_io,
{boost::asio::ip::address_v4::loopback(), 0}
};
epee::net_utils::blocked_mode_client client;
client.set_ssl(epee::net_utils::ssl_options_t{
epee::net_utils::ssl_support_t::e_ssl_support_enabled
});
const std::string port = std::to_string(acceptor.local_endpoint().port());
std::thread stopper{[&client] {
std::this_thread::sleep_for(std::chrono::milliseconds{100});
EXPECT_TRUE(client.shutdown());
}};
const auto start = std::chrono::steady_clock::now();
EXPECT_FALSE(client.connect("127.0.0.1", port, std::chrono::seconds{30}));
EXPECT_LT(std::chrono::steady_clock::now() - start, std::chrono::seconds{10});
stopper.join();
}
TEST(blocked_mode_client, shutdown_is_permanent)
{
boost::asio::io_context server_io;
boost::asio::ip::tcp::acceptor acceptor{
server_io,
{boost::asio::ip::address_v4::loopback(), 0}
};
epee::net_utils::blocked_mode_client client;
client.set_ssl(epee::net_utils::ssl_options_t{
epee::net_utils::ssl_support_t::e_ssl_support_disabled
});
const std::string port = std::to_string(acceptor.local_endpoint().port());
ASSERT_TRUE(client.connect("127.0.0.1", port, std::chrono::seconds{5}));
boost::system::error_code error;
boost::asio::ip::tcp::socket peer{server_io};
acceptor.accept(peer, error);
ASSERT_FALSE(error);
// the abort is sticky: a shutdown with nothing in flight still fails the next
// op promptly, so a request scheduled before it can never start a doomed call
EXPECT_TRUE(client.shutdown());
EXPECT_FALSE(client.is_connected());
std::string response;
const auto start = std::chrono::steady_clock::now();
EXPECT_FALSE(client.recv(response, std::chrono::seconds{30}));
EXPECT_LT(std::chrono::steady_clock::now() - start, std::chrono::seconds{10});
// the torn-down client never reconnects
EXPECT_FALSE(client.connect("127.0.0.1", port, std::chrono::seconds{5}));
}
TEST(tor_address, constants)
{
static_assert(!net::tor_address::is_local(), "bad is_local() response");