From 70ddaf8e667391103726049f25954aa0e3f80554 Mon Sep 17 00:00:00 2001 From: Lee *!* Clagett Date: Sat, 23 May 2026 01:02:42 -0400 Subject: [PATCH] /feed (websocket) implementation, with fallback to poll interface (#8) --- src/backend.cpp | 781 ++++++++++++++++++++++++++++--- src/backend.h | 79 +++- src/error.cpp | 6 + src/error.h | 3 + src/lwsf_config.h | 1 + src/lwsf_rpc.cpp | 161 ++++++- src/lwsf_rpc.h | 260 ++++++++++- src/net/CMakeLists.txt | 4 +- src/net/http.cpp | 92 +++- src/net/http.h | 18 +- src/net/websocket.cpp | 418 +++++++++++++++++ src/net/websocket.h | 85 ++++ src/pending_transaction.cpp | 6 +- src/wallet.cpp | 69 ++- src/wallet.h | 6 +- tests/unit/backend.test.cpp | 842 +++++++++++++++++++++++++++++++++- tests/unit/net/CMakeLists.txt | 2 +- tests/unit/net/push.test.cpp | 149 ++++++ tests/unit/net/push.test.h | 60 +++ 19 files changed, 2889 insertions(+), 153 deletions(-) create mode 100644 src/net/websocket.cpp create mode 100644 src/net/websocket.h create mode 100644 tests/unit/net/push.test.cpp create mode 100644 tests/unit/net/push.test.h diff --git a/src/backend.cpp b/src/backend.cpp index bdd2131..9869872 100644 --- a/src/backend.cpp +++ b/src/backend.cpp @@ -32,6 +32,8 @@ #include #include #include +#include + #include "crypto/crypto.h" // monero/src #include "crypto/crypto-ops.h" // monero/src #include "cryptonote_basic/cryptonote_basic_impl.h" // monero/src @@ -39,6 +41,7 @@ #include "error.h" #include "hex.h" // monero/src #include "lwsf_config.h" +#include "net/websocket.h" #include "ringct/rctOps.h" #include "wire.h" #include "wire/adapted/crypto.h" @@ -387,6 +390,31 @@ namespace lwsf { namespace internal { namespace backend return self.subaccounts[sub.maj_i].last < sub.min_i; // last is inclusive } + boost::optional is_real_spend(const account& self, const rpc::transaction_spend& source) + { + crypto::key_derivation derivation{}; + if (!crypto::generate_key_derivation(source.tx_pub_key, self.view.sec, derivation)) + return boost::none; + + crypto::public_key spend_pub{}; + const crypto::secret_key spend_sec = get_spend_secret(self, source.sender); + if (!crypto::secret_key_to_public_key(spend_sec, spend_pub)) + return boost::none; + + crypto::public_key output_pub{}; + if (!crypto::derive_public_key(derivation, source.out_index, spend_pub, output_pub)) + return boost::none; + + crypto::secret_key output_secret{}; + crypto::derive_secret_key(derivation, source.out_index, spend_sec, output_secret); + + crypto::key_image image{}; + crypto::generate_key_image(output_pub, output_secret, image); + if (source.key_image == image) + return output_pub; + return boost::none; + } + boost::optional> update_tx(const account& self, transaction& out, const rpc::transaction& source) { /* Let `receives` and `spends` re-populate in `merge_output` and @@ -412,29 +440,12 @@ namespace lwsf { namespace internal { namespace backend std::uint64_t total_spent = 0; for (const auto& spend : source.spent_outputs) { - crypto::key_derivation derivation{}; - if (!crypto::generate_key_derivation(spend.tx_pub_key, self.view.sec, derivation)) - continue; - - crypto::public_key spend_pub{}; - const crypto::secret_key spend_sec = get_spend_secret(self, spend.sender); - if (!crypto::secret_key_to_public_key(spend_sec, spend_pub)) - continue; - - crypto::public_key output_pub{}; - if (!crypto::derive_public_key(derivation, spend.out_index, spend_pub, output_pub)) - continue; - - crypto::secret_key output_secret{}; - crypto::derive_secret_key(derivation, spend.out_index, spend_sec, output_secret); - - crypto::key_image image{}; - crypto::generate_key_image(output_pub, output_secret, image); - if (image == spend.key_image) + const auto output_pub = is_real_spend(self, spend); + if (output_pub) { /* Frontend will typically know about spend before backend. So only merge and never erase spends. */ - const rpc::address_meta sub = update_spend(out.spends.try_emplace(image).first->second, spend, output_pub); + const rpc::address_meta sub = update_spend(out.spends.try_emplace(spend.key_image).first->second, spend, *output_pub); if (need_expansion(self, sub)) meta.push_back(sub); total_spent += std::uint64_t(spend.amount); @@ -517,12 +528,9 @@ namespace lwsf { namespace internal { namespace backend { const bool rescanning = source.scanned_block_height < tx.second->height.value_or(0); - if (rescanning || !tx.second->description.empty() || !tx.second->transfers.empty()) + if (rescanning || tx.second->has_metadata()) { - const auto iter = updated_txes.emplace_hint( - updated_txes.end(), tx.first, nullptr - ); - + const auto iter = updated_txes.emplace(tx.first, nullptr).first; if (!iter->second) { iter->second = std::make_shared(*tx.second); @@ -610,15 +618,7 @@ namespace lwsf { namespace internal { namespace backend if (self.primary.restore_height <= self.primary.requested_start && !source.lookahead_fail) self.import_error = {}; - self.fee_mask = unspents.fee_mask; - if (unspents.fees.empty()) - { - self.per_byte_fee.resize(1); - self.per_byte_fee[0] = unspents.per_byte_fee; - } - else - self.per_byte_fee = std::move(unspents.fees); - + self.update_fees(unspents); return out; } @@ -709,6 +709,447 @@ namespace lwsf { namespace internal { namespace backend runner(); } + void notify_listener(Monero::WalletListener& listener, const std::uint64_t new_height, const std::uint64_t orig_height, const merge_results& merged) + { + listener.refreshed(); + if (!merged.new_transactions.empty() || new_height - orig_height) + listener.updated(); + + for (std::uint64_t i = orig_height; i < new_height; ++i) + listener.newBlock(i); + + for (const auto& tx : merged.new_transactions) + { + const auto txid = epee::string_tools::pod_to_hex(tx->id); + if (tx->direction == Monero::TransactionInfo::Direction_In) + { + if (tx->height) + listener.moneyReceived(txid, tx->amount); + else + listener.unconfirmedMoneyReceived(txid, tx->amount); + } + else + listener.moneySpent(txid, tx->amount); + } + } + + class feed_loop + { + struct frame + { + std::weak_ptr self; + std::weak_ptr ws; + merge_results* merged; + std::function on_sync; + + explicit frame(std::shared_ptr in, merge_results* merged, std::function&& on_sync) + : self(in), ws(in->feed), merged(merged), on_sync(std::move(on_sync)) + {} + }; + + std::shared_ptr frame_; + + /* Convert the simplified '/feed:tx_sync:' message to a split + '/get_address_txs' and '/get_unspent_outs' format. This way both legacy + and new syncing mechanisms use the same code. This slows down the new + sync, but keeps the legacy sync fast, which occurs more frequently. */ + static expect> convert_to_legacy(rpc::feed_tx_sync src) + { + auto out = std::make_pair(rpc::get_address_txs{}, rpc::get_unspent_outs_response{}); + + out.first.lookahead_fail = src.lookahead_fail; + out.first.scanned_block_height = src.scanned_block_height; + out.first.start_height = src.start_height; + out.first.blockchain_height = src.blockchain_height; + out.first.lookahead = src.lookahead; + + std::sort( + src.transactions.begin(), src.transactions.end(), + [] (const auto& lhs, const auto& rhs) + { + static_assert(std::is_integral(), "optional types sort incorrectly"); + return lhs.height < rhs.height; + } + ); + + std::map receives; + for (const auto& orig : src.transactions) + { + for (const auto& receive : orig.receives) + { + receives.emplace(receive.id, receive); + out.second.outputs.emplace_back(receive, orig); + } + + auto& tx = out.first.transactions.emplace_back(orig); + for (const auto& orig_spend : orig.spends) + { + const auto receive = receives.find(orig_spend.id); + if (receive == receives.end()) + return {error::feed_missing_output}; + tx.spent_outputs.emplace_back(orig_spend, receive->second); + } + } + + return out; + } + + void on_error(wallet& self, boost::unique_lock& lock, const std::error_code error) + { + std::shared_ptr feed; + std::function on_sync; + + LWSF_VERIFY(lock.owns_lock()); + const std::function on_refresh{self.on_feed_refresh}; + feed.swap(self.feed); + on_sync.swap(frame_->on_sync); + + { + const boost::lock_guard lock_listener{self.sync_listener}; + lock.unlock(); + if (self.listener) + self.listener->refreshed(); + } + + MERROR("Error on websocket '/feed': " << error.message()); + if (feed) + websocket::async_close(std::move(feed)); + + if (on_sync) + on_sync(self.refresh_error = error); + else + backend::wallet::refresh(frame_->self.lock(), true, [] (auto) {}); + + if (on_refresh) + on_refresh(error); + } + + static std::string_view get_prefix(const epee::byte_slice& msg) noexcept + { + void const* const sep = std::memchr(msg.data(), u8':', msg.size()); + if (!sep) + return {}; + return { + reinterpret_cast(msg.data()), + std::size_t(reinterpret_cast(sep) - msg.data() + 1) + }; + } + + template + static expect unpack(epee::byte_slice&& msg) + { + const std::string_view prefix = get_prefix(msg); + if (prefix != T::prefix()) + return {error::bad_feed_prefix}; + msg.remove_prefix(prefix.size()); + + T out{}; + const std::error_code error = wire::msgpack::from_bytes(std::move(msg), out); + if (error) + return error; + return out; + } + + bool process(wallet& self, expect&& msg, boost::unique_lock& lock, merge_results& merged) + { + if (!msg) + { + on_error(self, lock, msg.error()); + return false; + } + + LWSF_VERIFY(lock.owns_lock()); + std::map> receives; + std::unordered_map> updated; + std::unordered_multimap> images; + updated.reserve(self.primary.txes.size() + msg->transactions.size()); + for (const auto& tx : self.primary.txes) + { + auto elem = updated.try_emplace(tx.first, std::make_shared(*tx.second)); + for (const auto& receive : elem.first->second->receives) + receives.emplace(receive.second.get_output_id(), elem.first->second); + for (const auto& spend : elem.first->second->spends) + images.emplace(spend.first, elem.first->second); + } + + for (const auto& new_tx : msg->transactions) + { + auto tx = updated.try_emplace(new_tx.hash, nullptr).first; + const bool is_new = !tx->second; + tx->second = std::make_shared(new_tx, tx->second.get()); + if (is_new) + merged.new_transactions.push_back(tx->second); + + std::uint64_t receiving = 0; + for (const auto& receive : new_tx.receives) + { + receiving += receive.amount; + receives.emplace(receive.id, tx->second); + if (need_expansion(self.primary, receive.recipient)) + merged.merge_subaddr(receive.recipient); + } + + std::uint64_t spending = 0; + for (const auto& spend : new_tx.spends) + { + const auto receive = receives.find(spend.id); + if (receive == receives.end()) + { + on_error(self, lock, error::feed_missing_output); + return false; + } + + std::pair const* output = nullptr; + for (const auto& existing : receive->second->receives) + { + if (existing.second.get_output_id() == spend.id) + { + output = std::addressof(existing); + break; + } + } + + if (output == nullptr) + { + on_error(self, lock, error::feed_missing_output); + return false; + } + + if (is_real_spend(self.primary, {spend, output->second.as_receive(output->first)})) + { + spending += output->second.amount; + tx->second->spends.try_emplace(spend.key_image, output->second, output->first); + auto failed = images.equal_range(spend.key_image); + for ( ; failed.first != failed.second; ++failed.first) + failed.first->second->failed = true; + } + } + + if (receiving || spending) + { + if (new_tx.fee <= spending) + spending -= new_tx.fee; + + if (tx->second->amount <= spending) + { + tx->second->direction = Monero::TransactionInfo::Direction_Out; + tx->second->amount = spending - tx->second->amount; + } + else + { + tx->second->direction = Monero::TransactionInfo::Direction_In; + tx->second->amount = tx->second->amount - spending; + } + } + else + updated.erase(new_tx.hash); // used solely as decoy in this tx + } + + self.primary.txes.swap(updated); + merged.lookahead_fail = msg->lookahead_fail; + self.server_lookahead = {msg->lookahead.maj_i, msg->lookahead.min_i}; + self.blockchain_height = std::max(self.blockchain_height, msg->blockchain_height); + self.primary.scan_height = std::max(self.primary.scan_height, msg->scan_end); + return true; + } + + void process(wallet& self, expect&& msg, boost::unique_lock& lock) + { + if (!msg) + return on_error(self, lock, msg.error()); + MERROR("Received websocket '/feed' error from server: " << msg->msg); + on_error(self, lock, error::feed); + } + + bool process(wallet& self, expect&& msg, boost::unique_lock& lock, merge_results& merged) + { + if (!msg) + { + on_error(self, lock, msg.error()); + return false; + } + + LWSF_VERIFY(lock.owns_lock()); + std::unordered_map> updated; + for (const auto& tx : self.primary.txes) + updated.emplace(tx.first, std::make_shared(*tx.second)); + + auto tx = updated.emplace(msg->hash, nullptr).first; + if (!tx->second) + { + tx->second = std::make_shared(*msg); + merged.new_transactions.push_back(tx->second); + if (need_expansion(self.primary, msg->recipient)) + merged.merge_subaddr(msg->recipient); + + auto receive = tx->second->receives.try_emplace(msg->public_key); + if (receive.second) + { + tx->second->amount += msg->amount; + const auto sub = update_output(receive.first->second, *msg, self.primary.view.sec); + if (need_expansion(self.primary, sub)) + merged.merge_subaddr(sub); + } + } // else we saw it because we spent it + + self.primary.txes.swap(updated); + return true; + } + + bool process(wallet& self, expect&& msg, boost::unique_lock& lock) + { + if (!msg) + { + on_error(self, lock, msg.error()); + return false; + } + + MDEBUG("Received websocket '/feed' warning from server: " << msg->msg); + if (msg->code != rpc::feed_status::blockchain_reorg) + return false; // skip notifying wallet + + LWSF_VERIFY(lock.owns_lock()); + std::unordered_map> updated; + for (const auto& tx : self.primary.txes) + { + LWSF_VERIFY(tx.second); + const bool by_height = + tx.second->height.value_or(std::numeric_limits::max()) <= msg->height; + if (by_height || tx.second->has_metadata()) + { + auto fresh = updated.emplace(tx.first, std::make_shared(*tx.second)); + if (!by_height) + { + fresh.first->second->timestamp = boost::none; + fresh.first->second->height = boost::none; + fresh.first->second->failed = false; + } + } + } + + self.primary.txes.swap(updated); + self.blockchain_height = std::min(self.blockchain_height, msg->height); + self.primary.restore_height = std::min(self.primary.restore_height, msg->height); + self.primary.scan_height = std::min(self.primary.scan_height, msg->height); + return true; + } + + public: + explicit feed_loop(std::shared_ptr self, merge_results* merged, std::function on_sync) + : frame_(std::make_shared(std::move(self), merged, std::move(on_sync))) + {} + + void operator()(std::error_code error = {}, copyable_slice incoming = {}) + { + LWSF_VERIFY(frame_); + const auto self_ptr = frame_->self.lock(); + LWSF_VERIFY(self_ptr); + wallet& self = *self_ptr; + + boost::unique_lock lock{self.sync}; + auto ws = frame_->ws.lock(); + if (ws.get() != self.feed.get()) // sketchy, relies on make_shared not freeing until weak_ptr closes + { + LWSF_VERIFY(!frame_->on_sync); // violation in `wallet::refresh` + if (ws) + websocket::async_close(std::move(ws)); + return; + } + + if (error) + return on_error(self, lock, error); + + self.last_sync = std::chrono::steady_clock::now(); + if (frame_->merged) + { + auto msg = unpack(std::move(incoming.data)); + if (!msg) + return on_error(self, lock, msg.error()); + + const auto split = convert_to_legacy(std::move(*msg)); + if (!split) + return on_error(self, lock, split.error()); + + const std::uint64_t orig_height = self.primary.scan_height; + msg = {error::feed}; // release memory + *frame_->merged = merge_response(self, split->first, split->second); + + std::function on_sync; + on_sync.swap(frame_->on_sync); + const std::function on_refresh{self.on_feed_refresh}; + + const std::uint64_t new_height = std::max(self.primary.scan_height, orig_height); + const boost::lock_guard listener_lock{self.sync_listener}; + lock.unlock(); + + if (self.listener) + notify_listener(*self.listener, new_height, orig_height, *frame_->merged); + + frame_->merged = nullptr; + if (on_sync) + on_sync({}); + if (on_refresh) + on_refresh({}); + } + else + { + if (incoming.data.empty()) + return on_error(self, lock, error::bad_feed_prefix); + + merge_results merged{}; + const std::uint64_t orig_height = self.primary.scan_height; + switch (incoming.data.data()[0]) + { + default: + return on_error(self, lock, error::bad_feed_prefix); + case 'b': + if (process(self, unpack(std::move(incoming.data)), lock, merged)) + break; + return; + case 'e': + return process(self, unpack(std::move(incoming.data)), lock); + case 'm': + if (process(self, unpack(std::move(incoming.data)), lock, merged)) + break; + return; + case 'w': + if (process(self, unpack(std::move(incoming.data)), lock)) + break; + return; + } + + const boost::lock_guard listener_lock{self.sync_listener}; + LWSF_VERIFY(lock.owns_lock()); + if (!self.listener && !self.on_feed_refresh) + return; + + const std::function on_refresh{self.on_feed_refresh}; + const std::uint64_t new_height = self.primary.scan_height; + + lock.unlock(); + if (self.listener) + notify_listener(*self.listener, new_height, orig_height, merged); + if (on_refresh) + on_refresh({}); + } + } + }; + + template + epee::byte_slice generate_feed_msg(const T& msg) + { + epee::byte_stream sink; + + const std::string_view prefix = T::prefix(); + sink.write(prefix.data(), prefix.size()); + + const std::error_code error = wire::msgpack::to_bytes(sink, msg); + if (error) + throw std::system_error{error, std::string{"Failed to create "} + T::prefix()}; + + return epee::byte_slice{std::move(sink)}; + } + //! Bind N arguments into a callback that takes zero arguments to run template struct binder @@ -733,14 +1174,14 @@ namespace lwsf { namespace internal { namespace backend F f_; template - void operator()(T... args) + void operator()(T... args) const { // Use `post` instead of `dispatch` due to locking in handlers const auto self = self_.lock(); LWSF_VERIFY(self); boost::asio::post( self->strand, - binder{self, std::move(f_), std::tuple(std::move(args)...)} + binder{self, f_, std::tuple(std::move(args)...)} ); } }; @@ -772,6 +1213,24 @@ namespace lwsf { namespace internal { namespace backend LWSF_VERIFY(self); boost::asio::post(self->strand, binder{self, std::move(f)}); } + + template + struct store_ + { + T* out; + F f; + + void operator()(const std::error_code error, T val) + { + LWSF_VERIFY(out); + *out = val; + f(error); + } + }; + + template + store_ store(T* out, F f) + { return {out, std::move(f)}; } } // anonymous WIRE_DEFINE_OBJECT(address_book_entry, map_address_book_entry); @@ -904,6 +1363,64 @@ namespace lwsf { namespace internal { namespace backend failed(false) {} + transaction::transaction(const rpc::feed_tx& src, const transaction* existing) + : raw_bytes(), + spends(), + receives(), + transfers(), + description(), + timestamp(src.timestamp), + height(src.get_height()), + amount(0), + fee(src.fee), + unlock_time(src.unlock_time), + direction(Monero::TransactionInfo::Direction_In), + payment_id(src.payment_id), + id(src.hash), + prefix(src.prefix_hash), + coinbase(src.coinbase), + failed(false) + { + for (const auto& receive : src.receives) + { + amount += receive.amount; + receives.try_emplace(receive.public_key, receive); + } + + if (existing) + { + transfers = existing->transfers; + description = existing->description; + } + } + + transaction::transaction(const rpc::feed_mempool& src) + : raw_bytes(), + spends(), + receives(), + transfers(), + description(), + timestamp(std::chrono::system_clock::now()), + height(), + amount(src.amount), + fee(src.fee), + unlock_time(src.unlock_time), + direction(Monero::TransactionInfo::Direction_In), + payment_id(src.payment_id), + id(src.hash), + prefix(src.prefix_hash), + coinbase(false), + failed(false) + { + auto elem = receives.try_emplace(src.public_key).first; + elem->second.global_index = std::numeric_limits::max(); + elem->second.amount = src.amount; + elem->second.recipient = src.recipient; + elem->second.index = src.index; + elem->second.rct_mask = src.rct; + elem->second.tx_pub = src.tx_pub_key; + } + transaction::transaction(const transaction& rhs) : raw_bytes(rhs.raw_bytes.clone()), // required because of this spends(rhs.spends), @@ -936,8 +1453,11 @@ namespace lwsf { namespace internal { namespace backend wallet::wallet(boost::asio::io_context& io) : listener(nullptr), + feed(nullptr), strand(io), client{}, + on_feed_refresh(nullptr), + on_feed_fail(nullptr), primary{}, per_byte_fee(), login_queue(), @@ -946,13 +1466,16 @@ namespace lwsf { namespace internal { namespace backend lookahead_error(default_subaddr_state), import_error(), last_sync(), + last_fees(), blockchain_height(0), fee_mask(0), server_lookahead{}, sync(), sync_listener(), sync_queue(), - passed_login(false) + passed_login(false), + probed_lookahead(false), + feed_disabled(false) { prep_primary_account(primary.subaccounts.emplace_back()); } @@ -963,6 +1486,11 @@ namespace lwsf { namespace internal { namespace backend { client.shutdown(); + const boost::lock_guard lock1{sync}; + feed = nullptr; + on_feed_refresh = nullptr; + on_feed_fail = nullptr; + const boost::lock_guard lock{sync_queue}; login_queue.clear(); refresh_queue.clear(); @@ -1017,6 +1545,24 @@ namespace lwsf { namespace internal { namespace backend primary.lookahead.minor <= server_lookahead.minor; } + bool wallet::need_fees() const noexcept + { + return config::refresh_interval_min <= std::chrono::steady_clock::now() - last_fees; + } + + void wallet::update_fees(const rpc::get_unspent_outs_response& unspents) + { + last_fees = std::chrono::steady_clock::now(); + fee_mask = unspents.fee_mask; + if (unspents.fees.empty()) + { + per_byte_fee.resize(1); + per_byte_fee[0] = unspents.per_byte_fee; + } + else + per_byte_fee = unspents.fees; + } + expect wallet::to_bytes() const { epee::byte_stream dest{}; @@ -1045,6 +1591,26 @@ namespace lwsf { namespace internal { namespace backend return status; } + void wallet::add_pending_tx(std::shared_ptr pending) + { + const auto txid = epee::string_tools::pod_to_hex(pending->id); + const std::uint64_t amount = pending->amount; + + { + const boost::lock_guard lock{sync}; + auto entry = primary.txes.try_emplace(pending->id).first; + if (!entry->second) + entry->second = std::move(pending); + } + const boost::lock_guard lock{sync_listener}; + if (!listener) + return; + + listener->refreshed(); + listener->updated(); + listener->moneySpent(txid, amount); + } + void wallet::login_is_new(std::shared_ptr self, std::function)> f) { struct frame @@ -1095,6 +1661,10 @@ namespace lwsf { namespace internal { namespace backend self.subaddress_error = {}; self.import_error = {}; self.lookahead_error = self.lookahead_good() ? std::error_code{} : error::subaddr_upgrade; + if (self.feed) + websocket::async_close(std::move(self.feed)); + self.feed = nullptr; + self.feed_disabled = false; frame_->login = rpc::login_request{ self.primary.address, self.primary.view.sec, self.primary.lookahead, true, self.primary.generated_locally @@ -1179,6 +1749,7 @@ namespace lwsf { namespace internal { namespace backend rpc::get_subaddrs subaddrs; merge_results merged; std::uint64_t orig_scan_height; + rpc::feed feed; const bool mandatory; explicit frame(std::shared_ptr in, const bool mandatory) @@ -1190,6 +1761,7 @@ namespace lwsf { namespace internal { namespace backend subaddrs{}, merged{}, orig_scan_height(0), + feed{}, mandatory(mandatory) {} @@ -1244,6 +1816,57 @@ namespace lwsf { namespace internal { namespace backend return self.run(self.refresh_queue, self.refresh_error = error); } + if (!self.feed_disabled) + { + if (!self.feed) + { + BOOST_ASIO_CORO_YIELD rpc::invoke_async( + self.client, rpc::empty{}, &frame_->feed, wrap(self_ptr, *this) + ); + if (error == http::error(501)) + { + ws_connect: + BOOST_ASIO_CORO_YIELD self.client.ws_async( + rpc::feed::endpoint(), rpc::feed::protocol(), store(&self.feed, wrap(self_ptr, *this)) + ); + + if (!error && self.feed) + { + // must hold lock to write `self.feed`, thus the frame + BOOST_ASIO_CORO_YIELD websocket::async_subscribe( + self.feed, + generate_feed_msg(rpc::feed_login{rpc::login{self.primary.address, self.primary.view.sec}}), + wrap(self_ptr, feed_loop{self_ptr, &frame_->merged, wrap(self_ptr, *this)}) + ); + if (!error) + goto txs_merged; + } + } + + if (self.feed) + websocket::async_close(std::move(self.feed)); + self.feed = nullptr; + self.feed_disabled = true; + { + std::function on_feed_fail; + std::function on_feed_refresh; + on_feed_fail.swap(self.on_feed_fail); + on_feed_refresh.swap(self.on_feed_refresh); + if (on_feed_fail) + on_feed_fail(error); + } + } + else // existing websocket stream to '/feed' + { + // ping should be quick, server doesn't hit disk, etc + BOOST_ASIO_CORO_YIELD websocket::async_ping(self.feed, wrap(self_ptr, *this)); + if (!error) + return self.run(self.refresh_queue, self.refresh_error); + self.feed = nullptr; + goto ws_connect; + } + } + frame_->orig_scan_height = self.primary.scan_height; frame_->login = rpc::login{self.primary.address, self.primary.view.sec}; BOOST_ASIO_CORO_YIELD rpc::invoke_async( @@ -1273,7 +1896,7 @@ namespace lwsf { namespace internal { namespace backend // Remember that this function provides the strong exception guarantee. frame_->merged = merge_response(self, frame_->txs_response, frame_->outs_response); - + txs_merged: if (!self.import_error && self.primary.requested_start < self.primary.restore_height) { BOOST_ASIO_CORO_YIELD restore_height( @@ -1342,29 +1965,10 @@ namespace lwsf { namespace internal { namespace backend std::max(frame_->orig_scan_height, self.primary.scan_height); lock.unlock(); - self.listener->refreshed(); - const auto& merged = frame_->merged; - if (!merged.new_transactions.empty() || new_scan_height - frame_->orig_scan_height) - self.listener->updated(); - - for (std::uint64_t i = frame_->orig_scan_height; i < new_scan_height; ++i) - self.listener->newBlock(i); - - for (const auto& tx : merged.new_transactions) - { - const auto txid = epee::string_tools::pod_to_hex(tx->id); - if (tx->direction == Monero::TransactionInfo::Direction_In) - { - if (tx->height) - self.listener->moneyReceived(txid, tx->amount); - else - self.listener->unconfirmedMoneyReceived(txid, tx->amount); - } - else - self.listener->moneySpent(txid, tx->amount); - } + notify_listener(*self.listener, new_scan_height, frame_->orig_scan_height, frame_->merged); self.run(self.refresh_queue, rc); + frame_->self.reset(); // prevent re-invoking listener } } }; @@ -1374,6 +1978,65 @@ namespace lwsf { namespace internal { namespace backend post(self, handler{std::make_shared(self, mandatory)}); } + void wallet::get_fees(std::shared_ptr self, std::function f) + { + struct frame + { + std::weak_ptr self; + rpc::get_unspent_outs_response response; + const std::function f; + + explicit frame(std::shared_ptr in, std::function&& f) + : self(std::move(in)), + response{}, + f(std::move(f)) + {} + }; + + struct handler : boost::asio::coroutine + { + std::shared_ptr frame_; + + explicit handler(std::shared_ptr in) + : boost::asio::coroutine(), frame_(std::move(in)) + {} + + void operator()(const std::error_code error = {}) + { + LWSF_VERIFY(frame_ && frame_->f); + if (error) + return frame_->f(error); + + const auto self_ptr = frame_->self.lock(); + LWSF_VERIFY(self_ptr); + wallet& self = *self_ptr; + assert(self.strand.running_in_this_thread()); + const boost::lock_guard lock{self.sync}; + BOOST_ASIO_CORO_REENTER(*this) + { + if (!self.need_fees()) + return frame_->f(error); + + if (!self.passed_login) + BOOST_ASIO_CORO_YIELD login(self_ptr, wrap(self_ptr, *this)); + + BOOST_ASIO_CORO_YIELD rpc::invoke_async( + self.client, + rpc::get_unspent_outs_request{rpc::login{self.primary.address, self.primary.view.sec}, rpc::uint64_string(0), 0, true}, + &frame_->response, + wrap(self_ptr, *this) + ); + + self.update_fees(frame_->response); + frame_->f(error); + } + } + }; + + LWSF_VERIFY(self && f); + post(self, handler{std::make_shared(self, std::move(f))}); + } + void wallet::register_subaccount(std::shared_ptr self, const std::uint32_t maj_i, std::function f) { struct frame diff --git a/src/backend.h b/src/backend.h index ee29348..4617c00 100644 --- a/src/backend.h +++ b/src/backend.h @@ -113,21 +113,8 @@ namespace lwsf { namespace internal { namespace backend }; WIRE_DECLARE_OBJECT(sub_account); - struct transfer_spend - { - std::uint64_t amount; - rpc::address_meta sender; - crypto::public_key tx_pub; - crypto::public_key output_pub; - - transfer_spend() noexcept - : amount(0), sender{}, tx_pub{}, output_pub{} - {} - }; - WIRE_DECLARE_OBJECT(transfer_spend); - struct transfer_in - { + { std::uint64_t global_index; std::uint64_t amount; rpc::address_meta recipient; @@ -143,9 +130,52 @@ namespace lwsf { namespace internal { namespace backend rct_mask(), tx_pub{} {} + + transfer_in(const rpc::feed_tx::receive& src) noexcept + : global_index(src.id.legacy.index), + amount(src.amount), + recipient(src.recipient), + index(src.index), + rct_mask(src.rct), + tx_pub(src.tx_pub_key) + {} + + rpc::feed_tx::output_id get_output_id() const noexcept + { + return { + rpc::feed_tx::legacy_id{ + bool(rct_mask) ? unsigned(0) : amount, + global_index + } + }; + } + + rpc::feed_tx::receive as_receive(const crypto::public_key& pub) const noexcept + { + return { + amount, get_output_id(), recipient, index, rct_mask, pub, tx_pub + }; + } }; WIRE_DECLARE_OBJECT(transfer_in); + struct transfer_spend + { + std::uint64_t amount; + rpc::address_meta sender; + crypto::public_key tx_pub; + crypto::public_key output_pub; + + transfer_spend() noexcept + : amount(0), sender{}, tx_pub{}, output_pub{} + {} + + transfer_spend(const transfer_in& src, const crypto::public_key& pub) noexcept + : amount(src.amount), sender(src.recipient), tx_pub(src.tx_pub), output_pub(pub) + {} + }; + WIRE_DECLARE_OBJECT(transfer_spend); + struct transfer_out { std::string address; @@ -165,12 +195,16 @@ namespace lwsf { namespace internal { namespace backend struct transaction { transaction(); + + transaction(const rpc::feed_tx& src, const transaction* existing = nullptr); + transaction(const rpc::feed_mempool& src); transaction(transaction&&) = default; transaction(const transaction& rhs); transaction& operator=(transaction&&) = default; bool is_unlocked(std::uint64_t chain_height, Monero::NetworkType type) const; + bool has_metadata() const noexcept { return !description.empty() || !transfers.empty(); } /*! flat_map is used here for faster copies/merging. A single allocation is needed in the copy (done every refresh interval), instead of an @@ -240,8 +274,11 @@ namespace lwsf { namespace internal { namespace backend // Everything uses `sync` for synchronization unless noted Monero::WalletListener* listener; //!< Use `sync_listener` + std::shared_ptr feed; boost::asio::io_context::strand strand; // on_feed_refresh; + std::function on_feed_fail; account primary; std::string client_prefix; std::vector per_byte_fee; //!< by priority level @@ -252,18 +289,21 @@ namespace lwsf { namespace internal { namespace backend std::error_code lookahead_error; //!< warnings/errors of `server_lookahead` value std::error_code import_error; //!< Error from `import_wallet_request` std::chrono::steady_clock::time_point last_sync; + std::chrono::steady_clock::time_point last_fees; std::uint64_t blockchain_height; std::uint64_t fee_mask; config::lookahead server_lookahead; //!< Status of lookahead server-side mutable boost::mutex sync; boost::mutex sync_listener; boost::mutex sync_queue; - bool passed_login; + bool passed_login; //!< Reset to false to re-initialize all other values bool probed_lookahead; //!< True iff queries were done to determine lookahead re-sync + bool feed_disabled; wallet(boost::asio::io_context& io); ~wallet() noexcept; + //! Extra guard against memory cycles void shutdown(); @@ -292,13 +332,14 @@ namespace lwsf { namespace internal { namespace backend // `sync` mutex is NOT acquired for this group - cryptonote::network_type get_net_type() const; crypto::public_key get_spend_public(const rpc::address_meta& index) const; cryptonote::account_keys get_primary_keys() const; cryptonote::account_public_address get_spend_account(const rpc::address_meta& index) const; std::string get_spend_address(const rpc::address_meta& index) const; bool lookahead_good() const noexcept; + bool need_fees() const noexcept; + void update_fees(const rpc::get_unspent_outs_response& unspents); // End GROUP @@ -310,6 +351,9 @@ namespace lwsf { namespace internal { namespace backend //! De-serialize `this` from msgpack. Locks+replaces contents. std::error_code from_bytes(epee::byte_slice source); + //! Do not hold `sync` or `sync_listener` + void add_pending_tx(std::shared_ptr pending); + // End GROUP // `sync` is acquired in a `asio::post` function - thread-safe and non-blocking until end @@ -329,6 +373,9 @@ namespace lwsf { namespace internal { namespace backend //! Refreshes txes information. Strong exception guarantee. static void refresh(std::shared_ptr self, bool mandatory, std::function); + //! Calls `get_unspent_outs` to update the fees when using websockets + static void get_fees(std::shared_ptr self, std::function); + //! Notify server that new major accounts need to be watched. static void register_subaccount(std::shared_ptr self, std::uint32_t maj_i, std::function f); diff --git a/src/error.cpp b/src/error.cpp index 1dc82fc..aab1c50 100644 --- a/src/error.cpp +++ b/src/error.cpp @@ -40,12 +40,18 @@ namespace lwsf return "No error"; case error::approval: return "Account needs approval on light-wallet-server"; + case error::bad_feed_prefix: + return "Websocket prefix on '/feed' message was invalid"; case error::configuration: return "Configuration issue"; case error::create: return "Account creation not possible"; case error::decryption: return "Unable to decrypt file (likely bad password)"; + case error::feed: + return "Websocket '/feed' had error requiring resync"; + case error::feed_missing_output: + return "Websocket '/feed' sent a spent without sending the receive"; case error::import_fee: return "Import/restore from height needs payment, see address book"; case error::import_invalid: diff --git a/src/error.h b/src/error.h index 901a9a2..d703b2b 100644 --- a/src/error.h +++ b/src/error.h @@ -43,9 +43,12 @@ namespace lwsf { none = 0, //!< Must be zero for `expect<..>` approval, //!< Account needs approval + bad_feed_prefix, //!< Websocket '/feed' received invalid prefix on msg configuration, //!< Bad Configuration create, //!< Account creation not possible decryption, //!< Failed to decrypt data + feed, //!< '/feed' websocket had error requiring restart + feed_missing_output,//!< '/feed' websocket sent spend without receive import_fee, //!< `import_wallet_request` has fee import_invalid, //!< `import_wallet_request` has invalid fee import_pending, //!< `import_wallet_request` not fulfilled/approved diff --git a/src/lwsf_config.h b/src/lwsf_config.h index a2bd9e9..d5e127d 100644 --- a/src/lwsf_config.h +++ b/src/lwsf_config.h @@ -58,6 +58,7 @@ namespace lwsf { namespace config constexpr const std::size_t max_txes_in_rpc = 2048; constexpr const std::size_t min_outputs = 2; constexpr const std::uint32_t mixin_default = 15; + constexpr const std::chrono::seconds ping_timeout{10}; constexpr const std::chrono::seconds refresh_interval{30}; constexpr const std::chrono::seconds refresh_interval_min{5}; constexpr const std::chrono::seconds rpc_timeout{30}; diff --git a/src/lwsf_rpc.cpp b/src/lwsf_rpc.cpp index 95a067a..4107610 100644 --- a/src/lwsf_rpc.cpp +++ b/src/lwsf_rpc.cpp @@ -34,6 +34,7 @@ #include #include #include +#include #include "hex.h" // monero/contrib/epee/include #include "lwsf_config.h" #include "ringct/rctOps.h" // monero/contrib/epee/include @@ -42,6 +43,7 @@ #include "wire/adapted/crypto.h" #include "wire/field.h" #include "wire/json.h" +#include "wire/msgpack.h" #include "wire/traits.h" #include "wire/wrapper/array.h" #include "wire/wrapper/defaulted.h" @@ -59,7 +61,7 @@ namespace lwsf { namespace internal { namespace rpc void read_bytes(wire::reader& source, empty&) { wire::object(source); } - void write_bytes(wire::json_writer& dest, const login& self) + void write_bytes(wire::writer& dest, const login& self) { wire::object(dest, WIRE_FIELD(address), WIRE_FIELD(view_key)); } @@ -92,9 +94,144 @@ namespace lwsf { namespace internal { namespace rpc WIRE_FIELD(height), WIRE_FIELD(target_height) ); - } + namespace + { + std::variant make_payment_id(const boost::optional& payment_id) + { + if (payment_id && !payment_id->empty()) + { + if (payment_id->size() == sizeof(crypto::hash8)) + { + crypto::hash8 out{}; + std::memcpy(std::addressof(out), payment_id->data(), sizeof(crypto::hash8)); + return out; + } + else if (payment_id->size() == sizeof(crypto::hash)) + { + crypto::hash out{}; + std::memcpy(std::addressof(out), payment_id->data(), sizeof(crypto::hash)); + return out; + } + else + WIRE_DLOG_THROW(wire::error::schema::fixed_binary, "Invalid payment_id size"); + } + + return empty{}; + } + } + + void read_bytes(wire::json_reader& source, const feed&) + { wire::object(source); } + + static void read_bytes(wire::msgpack_reader& source, feed_tx::legacy_id& self) + { wire::object(source, WIRE_FIELD(amount), WIRE_FIELD(index)); } + + static void read_bytes(wire::msgpack_reader& source, feed_tx::output_id& self) + { wire::object(source, WIRE_FIELD(legacy)); } + + static void read_bytes(wire::msgpack_reader& source, feed_tx::receive& self) + { + static constexpr std::uint64_t max = std::numeric_limits::max(); + static constexpr auto txpool = feed_tx::output_id{feed_tx::legacy_id{max, max}}; + wire::object(source, + WIRE_FIELD(amount), + WIRE_FIELD_DEFAULTED(id, txpool), + WIRE_FIELD_DEFAULTED(recipient, address_meta{}), + WIRE_FIELD(index), + WIRE_OPTIONAL_FIELD(rct), + WIRE_FIELD(public_key), + WIRE_FIELD(tx_pub_key) + ); + } + + static void read_bytes(wire::msgpack_reader& source, feed_tx::spend& self) + { wire::object(source, WIRE_FIELD(id), WIRE_FIELD(key_image)); } + + void read_bytes(wire::msgpack_reader& source, feed_tx& self) + { + using min_receive = wire::min_element_sizeof; + using min_spend = wire::min_element_sizeof; + using time_point = std::chrono::system_clock::time_point; + + time_point::rep timestamp = 0; + boost::optional payment_id; + + wire::object(source, + WIRE_FIELD_ARRAY(receives, min_receive), + WIRE_FIELD_ARRAY(spends, min_spend), + WIRE_FIELD(fee), + WIRE_FIELD(unlock_time), + WIRE_FIELD_DEFAULTED(height, feed_tx::txpool()), + wire::field("timestamp", std::ref(timestamp)), + WIRE_FIELD_DEFAULTED(mixin, std::numeric_limits::max()), + wire::optional_field("payment_id", std::ref(payment_id)), + WIRE_FIELD(hash), + WIRE_FIELD(prefix_hash), + WIRE_FIELD_DEFAULTED(coinbase, false), + WIRE_FIELD_DEFAULTED(mempool, false) + ); + + self.payment_id = make_payment_id(payment_id); + self.timestamp = time_point{std::chrono::seconds{timestamp}}; + } + + using min_tx_size = wire::min_element_sizeof; + void read_bytes(wire::msgpack_reader& source, feed_blocks& self) + { + wire::object(source, + WIRE_FIELD_ARRAY(transactions, min_tx_size), + WIRE_FIELD(scan_start), + WIRE_FIELD(scan_end), + WIRE_FIELD(blockchain_height), + WIRE_OPTIONAL_FIELD(lookahead_fail), + WIRE_FIELD_DEFAULTED(lookahead, address_meta{}) + ); + } + + void read_bytes(wire::msgpack_reader& source, feed_error& self) + { wire::object(source, WIRE_FIELD(msg), WIRE_FIELD(code)); } + + void write_bytes(wire::msgpack_writer& dest, const feed_login& self) + { wire::object(dest, WIRE_FIELD(account)); } + + void read_bytes(wire::msgpack_reader& source, feed_mempool& self) + { + boost::optional payment_id; + wire::object(source, + WIRE_FIELD(amount), + WIRE_FIELD(fee), + WIRE_FIELD(unlock_time), + WIRE_FIELD_DEFAULTED(recipient, address_meta{}), + WIRE_FIELD_DEFAULTED(mixin, std::numeric_limits::max()), + wire::optional_field("payment_id", std::ref(payment_id)), + WIRE_OPTIONAL_FIELD(rct), + WIRE_FIELD(hash), + WIRE_FIELD(prefix_hash), + WIRE_FIELD(public_key), + WIRE_FIELD(tx_pub_key), + WIRE_FIELD(index) + ); + + self.payment_id = make_payment_id(payment_id); + } + + void read_bytes(wire::msgpack_reader& source, feed_tx_sync& self) + { + wire::object(source, + WIRE_FIELD_ARRAY(transactions, min_tx_size), + WIRE_FIELD(scanned_block_height), + WIRE_FIELD(start_height), + WIRE_FIELD(blockchain_height), + WIRE_OPTIONAL_FIELD(lookahead_fail), + WIRE_FIELD_DEFAULTED(lookahead, address_meta{}) + ); + } + + void read_bytes(wire::msgpack_reader& source, feed_warning& self) + { wire::object(source, WIRE_FIELD(msg), WIRE_FIELD(height), WIRE_FIELD(code)); } + void write_bytes(wire::json_writer& dest, const uint64_string source) { const auto as_string = dest.to_string(std::uintmax_t(source)); @@ -147,26 +284,9 @@ namespace lwsf { namespace internal { namespace rpc WIRE_FIELD(mempool) ); - if (payment_id && !payment_id->empty()) - { - if (payment_id->size() == sizeof(crypto::hash8)) - { - self.payment_id = crypto::hash8{}; - std::memcpy(std::addressof(std::get(self.payment_id)), payment_id->data(), sizeof(crypto::hash8)); - } - else if (payment_id->size() == sizeof(crypto::hash)) - { - self.payment_id = crypto::hash{}; - std::memcpy(std::addressof(std::get(self.payment_id)), payment_id->data(), sizeof(crypto::hash)); - } - else - WIRE_DLOG_THROW(wire::error::schema::fixed_binary, "Invalid payment_id size"); - } - else - self.payment_id = empty{}; + self.payment_id = make_payment_id(payment_id); self.timestamp.reset(); - if (timestamp) { // %Y-%m-%dT%H:%M:%SZ @@ -349,7 +469,6 @@ namespace lwsf { namespace internal { namespace rpc wire::object(source, WIRE_FIELD_ARRAY(all_subaddrs, max_subaddrs)); } - namespace { // This field is a mess between implementations, this cleans it up a bit diff --git a/src/lwsf_rpc.h b/src/lwsf_rpc.h index c0dcf02..5049686 100644 --- a/src/lwsf_rpc.h +++ b/src/lwsf_rpc.h @@ -34,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -49,6 +50,7 @@ #include "wire/basic_value.h" #include "wire/fwd.h" #include "wire/json.h" +#include "wire/msgpack/fwd.h" #include "wire/traits.h" namespace lwsf { namespace internal { namespace rpc @@ -140,7 +142,7 @@ namespace lwsf { namespace internal { namespace rpc std::string address; crypto::secret_key view_key; }; - void write_bytes(wire::json_writer&, const login&); + void write_bytes(wire::writer&, const login&); struct login_request @@ -179,6 +181,197 @@ namespace lwsf { namespace internal { namespace rpc void read_bytes(wire::json_reader&, daemon_status&); + enum class feed_status : std::uint16_t + { + unspecified_error = 0, + account_not_found, + bad_address, + bad_view_key, + blockchain_reorg, + daemon_unresponsive, + parse_error, + protocol_error, + queue_error, + schema_error + }; + WIRE_AS_INTEGER(feed_status); + + struct feed + { + feed() = delete; + static constexpr const char* endpoint() noexcept { return "/feed"; } + static constexpr const char* protocol() noexcept { return "lws.feed.v0.msgpack"; } + }; + void read_bytes(wire::json_reader&, const feed&); + + struct feed_tx + { + struct legacy_id + { + legacy_id() = delete; + std::uint64_t amount; + std::uint64_t index; + }; + + struct output_id { legacy_id legacy; }; + + struct receive + { + std::uint64_t amount; + output_id id; + address_meta recipient; + std::uint16_t index; + boost::optional rct; + crypto::public_key public_key; + crypto::public_key tx_pub_key; + + receive() noexcept + : amount(0), id{}, recipient{}, index(0), rct(), public_key{}, tx_pub_key{} + {} + + receive(std::uint64_t amount, const output_id& id, const address_meta& recipient, std::uint16_t index, const boost::optional rct, const crypto::public_key& public_key, const crypto::public_key& tx_pub_key) noexcept + : amount(amount), id(id), recipient(recipient), index(index), rct(rct), public_key(public_key), tx_pub_key(tx_pub_key) + {} + }; + + struct spend + { + output_id id; + crypto::key_image key_image; + + spend() noexcept + : id{}, key_image{} + {} + }; + + std::vector receives; + std::vector spends; + std::uint64_t fee; + std::uint64_t unlock_time; + std::uint64_t height; + std::chrono::system_clock::time_point timestamp; + std::uint32_t mixin; + std::variant payment_id; + crypto::hash hash; + crypto::hash prefix_hash; + bool coinbase; + bool mempool; + + //! Hack because `boost/std::optional` sort left adjacent + static constexpr std::uint64_t txpool() noexcept { return std::uint64_t(-1); } + + feed_tx() + : receives(), + spends(), + unlock_time(0), + height(), + timestamp(), + mixin(), + payment_id(), + hash{}, + prefix_hash{}, + coinbase(false), + mempool(false) + {} + + boost::optional get_height() const noexcept + { return boost::make_optional(height != txpool(), height); } + }; + void read_bytes(wire::msgpack_reader&, feed_tx&); + + inline constexpr bool operator==(const feed_tx::output_id& lhs, const feed_tx::output_id& rhs) noexcept + { + return lhs.legacy.index == rhs.legacy.index && lhs.legacy.amount == rhs.legacy.amount; + } + inline constexpr bool operator!=(const feed_tx::output_id& lhs, const feed_tx::output_id& rhs) noexcept + { + return lhs.legacy.index != rhs.legacy.index || lhs.legacy.amount != rhs.legacy.amount; + } + inline constexpr bool operator<(const feed_tx::output_id& lhs, const feed_tx::output_id& rhs) noexcept + { + return lhs.legacy.amount == rhs.legacy.amount ? + lhs.legacy.index < rhs.legacy.index : lhs.legacy.amount < rhs.legacy.amount; + } + + struct feed_blocks + { + feed_blocks() = delete; + static constexpr const char* prefix() noexcept { return "blocks:"; } + + std::vector transactions; + std::uint64_t scan_start; + std::uint64_t scan_end; + std::uint64_t blockchain_height; + boost::optional lookahead_fail; + address_meta lookahead; + }; + void read_bytes(wire::msgpack_reader&, feed_blocks&); + + struct feed_error + { + feed_error() = delete; + static constexpr const char* prefix() noexcept { return "error:"; } + + std::string msg; + feed_status code; + }; + void read_bytes(wire::msgpack_reader&, feed_error&); + + struct feed_login + { + feed_login() = delete; + static constexpr const char* prefix() noexcept { return "login:"; } + login account; + }; + void write_bytes(wire::msgpack_writer&, const feed_login&); + + //! One specific output is sent per-message + struct feed_mempool + { + feed_mempool() = delete; + static constexpr const char* prefix() noexcept { return "mempool:"; } + + std::uint64_t amount; + std::uint64_t fee; + std::uint64_t unlock_time; + address_meta recipient; + std::uint32_t mixin; + std::variant payment_id; + boost::optional rct; + crypto::hash hash; + crypto::hash prefix_hash; + crypto::public_key public_key; + crypto::public_key tx_pub_key; + std::uint16_t index; + }; + void read_bytes(wire::msgpack_reader&, feed_mempool&); + + struct feed_tx_sync + { + feed_tx_sync() = delete; + static constexpr const char* prefix() noexcept { return "tx_sync:"; } + + std::vector transactions; + std::uint64_t scanned_block_height; + std::uint64_t start_height; + std::uint64_t blockchain_height; + boost::optional lookahead_fail; + address_meta lookahead; + }; + void read_bytes(wire::msgpack_reader&, feed_tx_sync&); + + struct feed_warning + { + feed_warning() = delete; + static constexpr const char* prefix() noexcept { return "warning:"; } + + std::string msg; + std::uint64_t height; + feed_status code; + }; + void read_bytes(wire::msgpack_reader&, feed_warning&); + + enum class uint64_string : std::uint64_t {}; void write_bytes(wire::json_writer&, uint64_string); void read_bytes(wire::json_reader&, uint64_string&); @@ -198,6 +391,14 @@ namespace lwsf { namespace internal { namespace rpc key_image{}, tx_pub_key{} {} + + transaction_spend(const feed_tx::spend& spend, const feed_tx::receive& receive) noexcept + : amount(uint64_string(receive.amount)), + sender(receive.recipient), + out_index(receive.index), + key_image(spend.key_image), + tx_pub_key(receive.tx_pub_key) + {} }; void read_bytes(wire::json_reader&, transaction_spend&); @@ -226,6 +427,24 @@ namespace lwsf { namespace internal { namespace rpc coinbase(false), mempool(false) {} + + transaction(const feed_tx& src) + : spent_outputs(), + payment_id(src.payment_id), + timestamp(std::chrono::system_clock::to_time_t(src.timestamp)), + fee(uint64_string(src.fee)), + total_received(uint64_string(0)), + unlock_time(src.unlock_time), + height(src.get_height()), + hash(src.hash), + coinbase(src.coinbase), + mempool(src.mempool) + { + std::uint64_t total = 0; + for (const auto& received : src.receives) + total += received.amount; + total_received = uint64_string(total); + } }; void read_bytes(wire::json_reader&, transaction&); @@ -333,8 +552,6 @@ namespace lwsf { namespace internal { namespace rpc struct ringct { - ringct() = delete; - enum class format : std::uint8_t { none = 0, //!< Not ringct @@ -343,6 +560,19 @@ namespace lwsf { namespace internal { namespace rpc unencrypted }; + ringct() noexcept + : mask{}, type(format::none) + {} + + ringct(const rct::key& mask, const format type) noexcept + : mask(mask), type(type) + {} + + ringct(const boost::optional& src) noexcept + : mask(src.value_or(rct::key{})), + type(bool(src) ? format::unencrypted : format::none) + {} + rct::key mask; format type; }; @@ -370,6 +600,30 @@ namespace lwsf { namespace internal { namespace rpc public_key{}, tx_pub_key{} {} + + output(const feed_mempool& src) noexcept + : amount(uint64_string(src.amount)), + global_index(uint64_string(std::numeric_limits::max())), + recipient(src.recipient), + index(src.index), + rct(src.rct), + tx_hash(src.hash), + tx_prefix_hash(src.prefix_hash), + public_key(src.public_key), + tx_pub_key(src.tx_pub_key) + {} + + output(const feed_tx::receive& src, const feed_tx& tx) noexcept + : amount(uint64_string(src.amount)), + global_index(uint64_string(src.id.legacy.index)), + recipient(src.recipient), + index(src.index), + rct(src.rct), + tx_hash(tx.hash), + tx_prefix_hash(tx.prefix_hash), + public_key(src.public_key), + tx_pub_key(src.tx_pub_key) + {} }; void read_bytes(wire::json_reader&, output&); diff --git a/src/net/CMakeLists.txt b/src/net/CMakeLists.txt index e5e7539..35b2cce 100644 --- a/src/net/CMakeLists.txt +++ b/src/net/CMakeLists.txt @@ -26,8 +26,8 @@ # STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF # THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -set(lwsf-net_sources context.cpp http.cpp wallet_io.cpp) -set(lwsf-net_headers context.h http.h wallet_io.h) +set(lwsf-net_sources context.cpp http.cpp wallet_io.cpp websocket.cpp) +set(lwsf-net_headers context.h http.h wallet_io.h websocket.h) add_library(lwsf-net ${lwsf-net_sources} ${lwsf-net_headers}) target_link_libraries(lwsf-net PRIVATE monero::libraries) diff --git a/src/net/http.cpp b/src/net/http.cpp index b253992..980ff88 100644 --- a/src/net/http.cpp +++ b/src/net/http.cpp @@ -58,6 +58,7 @@ #include "misc_log_ex.h" // monero/contrib/epee/include #include "net/net_utils_base.h" #include "net/socks.h" // monero/src +#include "net/websocket.h" #include "string_tools.h" // monero/contrib/epee/include namespace lwsf { namespace internal { namespace http @@ -211,7 +212,7 @@ namespace lwsf { namespace internal { namespace http void(std::shared_ptr, std::function); boost::beast::flat_static_buffer buffer; - std::function connect; + const std::function connect; std::deque outgoing; const std::string host; const std::string prefix; @@ -435,6 +436,14 @@ namespace lwsf { namespace internal { namespace http self.is_connected = true; while (!self.outgoing.empty()) { + if (self.outgoing.front().verb == boost::beast::http::verb::subscribe) + { + std::function notifier; + notifier.swap(self.outgoing.front().notifier); + self.outgoing.clear(); + return notifier({}, {}); + } + set_timeout(self_, config::rpc_timeout); MDEBUG("Sending " << self.outgoing.front().body.size() << " bytes in HTTP " << (self.outgoing.front().is_get() ? "GET" : "POST") << " to " << self.outgoing.front()); @@ -490,31 +499,35 @@ namespace lwsf { namespace internal { namespace http } } }; + + void queue_async_(std::shared_ptr state, std::string&& target, epee::byte_slice&& body, std::function&& notifier, boost::beast::http::verb verb) + { + message msg{std::move(body), std::move(target), std::move(notifier), verb}; + if (!state) + return msg.notifier(common_error::kInvalidArgument, epee::byte_slice{}); + + MDEBUG("Queueing HTTP " << to_string(verb) << " to " << msg << " using " << state.get()); + boost::asio::dispatch( + state->strand, + [state, msg = std::move(msg)] () mutable + { + const bool empty = state->outgoing.empty(); + state->outgoing.push_back(std::move(msg)); + if (empty) + boost::asio::post(state->strand, client_loop{state}); + } + ); + } } // anonymous void client::queue_async(std::string&& target, epee::byte_slice&& body, std::function&& notifier, boost::beast::http::verb verb) const { - message msg{std::move(body), std::move(target), std::move(notifier), verb}; std::shared_ptr state; { const boost::lock_guard lock{sync_}; state = state_; } - - if (!state) - return msg.notifier(common_error::kInvalidArgument, epee::byte_slice{}); - - MDEBUG("Queueing HTTP " << (msg.is_get() ? "GET" : "POST") << " to " << msg << " using " << this); - boost::asio::dispatch( - state->strand, - [state, msg = std::move(msg)] () mutable - { - const bool empty = state->outgoing.empty(); - state->outgoing.push_back(std::move(msg)); - if (empty) - boost::asio::post(state->strand, client_loop{state}); - } - ); + queue_async_(std::move(state), std::move(target), std::move(body), std::move(notifier), verb); } void client::direct::operator()(std::shared_ptr self, std::function f) const @@ -685,6 +698,51 @@ namespace lwsf { namespace internal { namespace http state_ = std::make_shared(state_->strand.context(), state_->host, state_->prefix, state_->port, state_->ssl, proxy_); } + void client::ws_async(std::string target, std::string protocol, std::function notifier) + { + std::shared_ptr self; + { + const boost::lock_guard lock{sync_}; + self = state_; + state_ = std::make_shared(self->strand.context(), self->host, self->prefix, self->port, self->ssl, proxy_); + } + LWSF_VERIFY(self); + + struct start_ws + { + std::weak_ptr self_; + std::string target_; + std::string protocol_; + std::function notifier_; + + void operator()(const std::error_code error, epee::byte_slice) + { + const auto self = self_.lock(); + LWSF_VERIFY(self && notifier_); + if (error) + return notifier_(error, nullptr); + + std::string host; + boost::system::error_code ec{}; + boost::asio::ip::make_address_v6(self->host, ec); + const bool https = self->ssl_status == epee::net_utils::ssl_support_t::e_ssl_support_enabled; + if (!ec) + host = "[" + host + "]:" + std::to_string(self->port); + else if ((https && self->port == 443) || (!https && self->port == 80)) + host = self->host; + else + host = self->host + ":" + std::to_string(self->port); + + if (self->ssl_status == epee::net_utils::ssl_support_t::e_ssl_support_enabled) + websocket::async_start(std::move(self->sock), self->strand.context(), std::move(host), std::move(target_), std::move(protocol_), std::move(notifier_)); + else + websocket::async_start(std::move(self->sock.next_layer()), self->strand.context(), std::move(host), std::move(target_), std::move(protocol_), std::move(notifier_)); + } + }; + + queue_async_(self, {}, {}, start_ws{self, std::move(target), std::move(protocol), std::move(notifier)}, boost::beast::http::verb::subscribe); + } + void client::post_async(std::string url, epee::byte_slice json_body, std::function notifier) const { return queue_async(std::move(url), std::move(json_body), std::move(notifier), boost::beast::http::verb::post); diff --git a/src/net/http.h b/src/net/http.h index 68f9ae4..581dfa5 100644 --- a/src/net/http.h +++ b/src/net/http.h @@ -40,7 +40,9 @@ #include "byte_slice.h" // monero/contrib/epee/include #include "net/net_ssl.h" // monero/contrib/epee/include -namespace lwsf { namespace internal { namespace http +namespace lwsf { namespace internal { +namespace websocket { class stream; } +namespace http { enum class error : int { @@ -57,6 +59,9 @@ namespace lwsf { namespace internal { namespace http using server_response_func = void(std::error_code, epee::byte_slice); using callback_func = void(boost::system::error_code); using connect_func = void(std::shared_ptr, std::function); + using ws_func = void(std::error_code, std::shared_ptr); + + std::string make_host(const std::string& host); //! Primarily for webhooks, where the response is (basically) ignored. class client @@ -107,6 +112,17 @@ namespace lwsf { namespace internal { namespace http `success()` is returned. \return `success()` if `url` is valid. */ void get_async(std::string target, std::function notifier) const; + + + /*! Never blocks. Thread safe. Makes a HTTP "upgrade" websocket request + using the existing HTTP connection. If the handshake succeeds, the + `websocket::stream` handle is sent to `notifier`, else the error is + reported by same function. + + Any existing HTTP connection is irrevocably converted to a websocket, + however `post_async` and `get_async` will create a new connection, when + needed, using the previously supplied values to `init` and `proxy`. */ + void ws_async(std::string target, std::string protocol, std::function notifier); }; }}} // lwsf // internal // http diff --git a/src/net/websocket.cpp b/src/net/websocket.cpp new file mode 100644 index 0000000..227d952 --- /dev/null +++ b/src/net/websocket.cpp @@ -0,0 +1,418 @@ +// Copyright (c) 2026, The Monero Project +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without modification, are +// permitted provided that the following conditions are met: +// +// 1. Redistributions of source code must retain the above copyright notice, this list of +// conditions and the following disclaimer. +// +// 2. Redistributions in binary form must reproduce the above copyright notice, this list +// of conditions and the following disclaimer in the documentation and/or other +// materials provided with the distribution. +// +// 3. Neither the name of the copyright holder nor the names of its contributors may be +// used to endorse or promote products derived from this software without specific +// prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY +// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF +// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL +// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, +// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF +// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#include "websocket.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "byte_slice.h" // monero/contrib/epee/include +#include "byte_stream.h" // monero/contrib/epee/include +#include "error.h" +#include "lwsf_config.h" + +namespace lwsf { namespace internal { namespace websocket +{ + using const_flat_buffer = boost::beast::net::const_buffer; + + struct read_loop + { + std::shared_ptr self_; + std::function receive_; + void operator()(boost::system::error_code, std::size_t); + }; + + struct subscription : boost::asio::coroutine + { + std::shared_ptr self_; + copyable_slice topic_; + std::function receive_; + + subscription(subscription&&) = default; + subscription(const subscription&) = default; + explicit subscription(std::shared_ptr self, epee::byte_slice&& topic, std::function&& receive) + : self_(std::move(self)), topic_(std::move(topic)), receive_(std::move(receive)) + {} + + void operator()(boost::system::error_code = {}); + }; + + class stream + { + std::vector> pings_; + boost::asio::io_context::strand strand_; + boost::asio::steady_timer timer_; + boost::beast::multi_buffer buffer_; + bool writing_; + bool closing_; + bool subscribed_; + + struct on_timeout + { + std::shared_ptr self_; + void operator()(const boost::system::error_code error) const + { + LWSF_VERIFY(self_); + if (!error || error != boost::asio::error::operation_aborted) + { + self_->notify_pings(boost::asio::error::operation_aborted); + self_->async_close(self_); + } + } + }; + + protected: + + template + auto bind(F&& f) -> decltype(boost::asio::bind_executor(strand_, std::forward(f))) + { return boost::asio::bind_executor(strand_, std::forward(f)); } + + template + void do_async_read(T& sock, F f) + { + sock.async_read(buffer_, bind(std::move(f))); + } + + template + void do_async_write(T& sock, const const_flat_buffer buf, F f) + { + if (closing_) + return f(boost::beast::websocket::error::closed); + + LWSF_VERIFY(f.self_); + if (!writing_) + { + struct on_write + { + F f_; + + void operator()(const boost::system::error_code error, std::size_t) + { + const std::shared_ptr self = f_.self_; + LWSF_VERIFY(self); + + self->timer_.cancel(); + self->writing_ = false; + f_(error); + if (error) + { + self->notify_pings(error); + self->async_close(self); + } + else if (!self->pings_.empty()) + self->async_ping(on_ping{self}); + } + }; + + writing_ = true; + timer_.expires_after(config::ping_timeout); // timeout for write not response + timer_.async_wait(on_timeout{f.self_}); + sock.async_write(buf, bind(on_write{std::move(f)})); + } + else + pings_.push_back(std::bind(&stream::async_write, f.self_, std::move(f))); + } + + template + void do_async_ping(T& sock, F&& f) + { + if (closing_) + return f(boost::beast::websocket::error::closed); + + LWSF_VERIFY(!writing_ && f.self_); + writing_ = true; + timer_.expires_after(config::ping_timeout); + timer_.async_wait(on_timeout{f.self_}); + sock.async_ping(boost::beast::websocket::ping_data(), bind(std::forward(f))); + } + + template + void do_async_close(T& sock, std::shared_ptr self) + { + LWSF_VERIFY(self); + if (!writing_) + { + const bool run = !closing_; + closing_ = true; + if (run) + sock.async_close(boost::beast::websocket::close_reason(), bind(std::bind(&stream::async_shutdown, self, self))); + } + else + pings_.push_back(std::bind(&stream::async_close, self, self)); + } + + void do_async_shutdown(tcp_sock& sock, std::shared_ptr) + { do_cleanup(sock); } + + void do_async_shutdown(boost::asio::ssl::stream& sock, std::shared_ptr self) + { + LWSF_VERIFY(self); + sock.async_shutdown(bind(std::bind(&stream::cleanup, self))); + } + + void do_cleanup(tcp_sock& sock) + { + boost::system::error_code ec{}; + sock.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec); + sock.close(ec); + } + + void do_cleanup(boost::asio::ssl::stream& sock) + { do_cleanup(sock.next_layer()); } + + public: + explicit stream(boost::asio::io_context& io) + : strand_(io), timer_(io), buffer_(), writing_(false), closing_(false), subscribed_(false) + {} + + //! \return True if this function has never been invoked previously + bool subscribe() noexcept + { + const bool rc = !subscribed_; + subscribed_ = true; + return rc; + } + + template + void dispatch(F&& f) + { + boost::asio::dispatch(strand_, std::forward(f)); + } + + epee::byte_slice gather_buffer() + { + epee::byte_stream sink{}; + sink.reserve(buffer_.size()); + + const auto buffers = buffer_.cdata(); + const auto end = boost::asio::buffer_sequence_end(buffers); + for (auto cur = boost::asio::buffer_sequence_begin(buffers); cur != end; ++cur) + { + const const_flat_buffer this_buf = *cur; + sink.write(reinterpret_cast(this_buf.data()), this_buf.size()); + } + + buffer_.consume(buffer_.size()); + return epee::byte_slice{std::move(sink)}; + } + + struct on_ping + { + std::shared_ptr self_; + void operator()(const boost::system::error_code& error) + { + LWSF_VERIFY(self_); + self_->writing_ = false; + if (self_->timer_.cancel() || error) + { + self_->notify_pings(error); + if (error) + self_->async_close(self_); + } + } + }; + + void notify_pings(const boost::system::error_code error = {}) + { + std::vector> pings; + pings.swap(pings_); + for (const auto& ping : pings) + ping(error); + } + + bool add_ping_notifier(std::function&& notifier) + { + pings_.push_back(std::move(notifier)); + return !writing_; + } + + virtual void async_read(read_loop&&) = 0; + virtual void async_write(subscription) = 0; + virtual void async_ping(on_ping&&) = 0; + virtual void async_close(std::shared_ptr) = 0; + virtual void async_shutdown(std::shared_ptr) = 0; + virtual void cleanup() = 0; + }; + + template + class stream_ final : public stream + { + boost::beast::websocket::stream sock_; + + public: + explicit stream_(T&& sock, boost::asio::io_context& io) + : stream(io), sock_(std::move(sock)) + { + sock_.binary(true); + } + + template + void async_handshake(const std::string& host, const std::string& target, std::string&& protocol, F f) + { + namespace http = boost::beast::http; + struct on_control + { + std::weak_ptr self_; + void operator()(boost::beast::websocket::frame_type, boost::beast::string_view) + { + const auto self = self_.lock(); + if (self) + self->notify_pings(); + } + }; + + LWSF_VERIFY(f.self_); + sock_.control_callback(on_control{f.self_}); + sock_.set_option( + boost::beast::websocket::stream_base::decorator( + [protocol = std::move(protocol)] (http::request_header<>& hdr) + { hdr.set(http::field::sec_websocket_protocol, protocol); } + ) + ); + sock_.async_handshake(host, target, bind(std::move(f))); + } + + virtual void async_read(read_loop&& handler) override final + { do_async_read(sock_, std::move(handler)); } + + virtual void async_write(subscription handler) override final + { + std::uint8_t const* const data = handler.topic_.data.data(); + const std::size_t size = handler.topic_.data.size(); + do_async_write(sock_, const_flat_buffer(data, size), std::move(handler)); + } + + virtual void async_ping(on_ping&& handler) override final + { do_async_ping(sock_, std::move(handler)); } + + virtual void async_close(std::shared_ptr self) override final + { do_async_close(sock_, std::move(self)); } + + virtual void async_shutdown(std::shared_ptr self) override final + { do_async_shutdown(sock_.next_layer(), std::move(self)); } + + virtual void cleanup() override final + { do_cleanup(sock_.next_layer()); } + }; + + struct after_connect + { + std::shared_ptr self_; + std::function notifier_; + + void operator()(const boost::system::error_code error) + { + LWSF_VERIFY(notifier_); + if (error) + notifier_(error, nullptr); + else + notifier_(error, self_); + } + }; + + void read_loop::operator()(const boost::system::error_code error, std::size_t) + { + LWSF_VERIFY(self_ && receive_); + stream& self = *self_; + receive_(error, self.gather_buffer()); + if (!error) + self.async_read(std::move(*this)); + else + self.async_close(self_); + } + + void subscription::operator()(boost::system::error_code error) + { + LWSF_VERIFY(self_ && receive_); + stream& self = *self_; + BOOST_ASIO_CORO_REENTER(*this) + { + if (self.subscribe()) + BOOST_ASIO_CORO_YIELD self.async_write(std::move(*this)); + else + error = boost::asio::error::already_started; + + if (error) + { + receive_(error, epee::byte_slice{}); + self.async_close(std::move(self_)); + } + else + self.async_read(read_loop{std::move(self_), std::move(receive_)}); + } + } + + template + inline void do_start(std::shared_ptr> self, const std::string& host, const std::string& target, std::string&& protocol, std::function&& notifier) + { + self->async_handshake(host, target, std::move(protocol), after_connect{self, std::move(notifier)}); + } + + void async_start(tcp_sock&& sock, boost::asio::io_context& io, std::string&& host, std::string&& target, std::string&& protocol, std::function&& notifier) + { + do_start(std::make_shared>(std::move(sock), io), host, target, std::move(protocol), std::move(notifier)); + } + + void async_start(boost::asio::ssl::stream&& sock, boost::asio::io_context& io, std::string&& host, std::string&& target, std::string&& protocol, std::function&& notifier) + { + do_start(std::make_shared>>(std::move(sock), io), host, target, std::move(protocol), std::move(notifier)); + } + + void async_subscribe(std::shared_ptr self, epee::byte_slice&& topic, std::function&& receive) + { + LWSF_VERIFY(self); + self->dispatch(subscription{self, std::move(topic), std::move(receive)}); + } + + void async_ping(std::shared_ptr self, std::function&& notifier) + { + LWSF_VERIFY(self); + self->dispatch([self, notifier = std::move(notifier)] () mutable { + if(self->add_ping_notifier(std::move(notifier))) + self->async_ping(stream::on_ping{self}); + }); + } + + void async_close(std::shared_ptr self) + { + LWSF_VERIFY(self); + self->dispatch(std::bind(&stream::async_close, self, self)); + } +}}} // lwsf // internal // websocket diff --git a/src/net/websocket.h b/src/net/websocket.h new file mode 100644 index 0000000..5948cc5 --- /dev/null +++ b/src/net/websocket.h @@ -0,0 +1,85 @@ +// Copyright (c) 2026, The Monero Project +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without modification, are +// permitted provided that the following conditions are met: +// +// 1. Redistributions of source code must retain the above copyright notice, this list of +// conditions and the following disclaimer. +// +// 2. Redistributions in binary form must reproduce the above copyright notice, this list +// of conditions and the following disclaimer in the documentation and/or other +// materials provided with the distribution. +// +// 3. Neither the name of the copyright holder nor the names of its contributors may be +// used to endorse or promote products derived from this software without specific +// prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY +// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF +// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL +// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, +// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF +// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#include "byte_slice.h" // monero/contrib/epee/include + +namespace lwsf { namespace internal +{ + + struct copyable_slice + { + copyable_slice() noexcept + : data(nullptr) + {} + + copyable_slice(epee::byte_slice&& data) noexcept + : data(std::move(data)) + {} + + copyable_slice(copyable_slice&&) noexcept = default; + copyable_slice(const copyable_slice& rhs) noexcept + : data(rhs.data.clone()) + {} + + epee::byte_slice data; + }; + + namespace websocket + { + class stream; + using tcp_sock = boost::asio::ip::tcp::socket; + using ready_func = void(std::error_code, std::shared_ptr); + using update_func = void(std::error_code, copyable_slice); + using ping_func = void(std::error_code); + + void async_start(tcp_sock&& sock, boost::asio::io_context& io, std::string&& host, std::string&& target, std::string&& protocol, std::function&& notifier); + void async_start(boost::asio::ssl::stream&& sock, boost::asio::io_context& io, std::string&& host, std::string&& target, std::string&& protocol, std::function&& notifier); + + /*! Subscribe to `topic` and call `notifier` on each incoming message. Can + only invoke once. Thread-safe. + \param topic The raw message to be sent to server + \param notifier Called on every received message. + \throw std::invalid_argument if invoked multiple times. */ + void async_subscribe(std::shared_ptr self, epee::byte_slice&& topic, std::function&& notifier); + + //! Ping the server at any time. Multiple calls allowed. Thread safe. + void async_ping(std::shared_ptr self, std::function&& notifier); + + //! Terminate the stream gracefully + void async_close(std::shared_ptr self); + } // websocket +}} // lwsf // internal diff --git a/src/pending_transaction.cpp b/src/pending_transaction.cpp index 9334aad..1af0cbd 100644 --- a/src/pending_transaction.cpp +++ b/src/pending_transaction.cpp @@ -155,10 +155,8 @@ namespace lwsf { namespace internal error_ = error.message(); return false; } - const boost::lock_guard lock{data_.wallet->sync}; - auto entry = data_.wallet->primary.txes.try_emplace(e->id).first; - if (!entry->second) - entry->second = std::move(e); + + data_.wallet->add_pending_tx(std::move(e)); } } catch(const std::exception& e) diff --git a/src/wallet.cpp b/src/wallet.cpp index 7070472..e323a6e 100644 --- a/src/wallet.cpp +++ b/src/wallet.cpp @@ -59,6 +59,7 @@ #include "net/http.h" #include "net/net_parse_helpers.h" // monero/contrib/epee/include #include "net/parse.h" // monero/src +#include "net/websocket.h" #include "numeric.h" #include "pending_transaction.h" #include "subaddress_account.h" @@ -304,6 +305,12 @@ namespace lwsf { namespace internal self->set_error(error, false); } + void wallet::frame::async_clear_error::operator()(const std::error_code error) const + { + LWSF_VERIFY(self); + self->set_error(error, true); + } + bool wallet::set_error(const std::error_code error, const bool clear) const { return status_->set_error(error, clear); @@ -366,10 +373,16 @@ namespace lwsf { namespace internal { void operator()(wallet* val) const { - if (!val) - return; - const boost::lock_guard lock{val->status_->sync_}; - val->status_->thread_state_ = state::stop; + LWSF_VERIFY(val && val->data_.wallet); + { + const boost::lock_guard lock{val->status_->sync_}; + val->status_->thread_state_ = state::stop; + } + { + const boost::lock_guard lock{val->data_.wallet->sync}; + val->data_.wallet->on_feed_refresh = nullptr; + val->data_.wallet->on_feed_fail = nullptr; + } } }; @@ -445,12 +458,22 @@ namespace lwsf { namespace internal const std::unique_ptr set_stop{this}; try { + LWSF_VERIFY(data_.wallet); + std::shared_ptr timer; boost::unique_lock lock{status_->sync_}; if (status_->thread_state_ != state::stop) { timer = std::make_shared(data_.context->io_); - backend::wallet::refresh(data_.wallet, false /* mandatory */, on_refresh{timer, status_, data_.wallet}); + const boost::lock_guard lock_wallet{data_.wallet->sync}; + if (!data_.wallet->feed_disabled) + { + data_.wallet->on_feed_fail = on_refresh{timer, status_, data_.wallet}; + data_.wallet->on_feed_refresh = frame::async_clear_error{status_}; + backend::wallet::refresh(data_.wallet, false /* mandatory */, frame::async_clear_error{status_}); + } + else + backend::wallet::refresh(data_.wallet, false /* mandatory */, on_refresh{timer, status_, data_.wallet}); } do { @@ -688,7 +711,12 @@ namespace lwsf { namespace internal status_->exception_error_ = "view_key, spend_key, and address_string do not match"; } - wallet::~wallet() { stop_refresh_loop(); } + wallet::~wallet() + { + stop_refresh_loop(); + if (data_.wallet) + data_.wallet->shutdown(); + } void wallet::add_subaddress(const std::uint32_t accountIndex, std::string label) { @@ -870,7 +898,7 @@ namespace lwsf { namespace internal return false; if (!daemon_username.empty() || !daemon_password.empty()) - throw std::runtime_error{"HTTP loging not supported"}; + throw std::runtime_error{"HTTP login not supported"}; // verify cert if `use_ssl == true`, otherwise autodetect if `https` // specified. @@ -909,9 +937,15 @@ namespace lwsf { namespace internal } } + // backend permanently assumes `/feed` failure even after changing + // http server const boost::unique_lock lock{data_.wallet->sync}; data_.wallet->passed_login = false; data_.wallet->client.init(data_.context->io_, std::move(url.host), std::move(url.uri), boost::numeric_cast(url.port), std::move(options)); + std::shared_ptr feed; + feed.swap(data_.wallet->feed); + if (feed) + websocket::async_close(std::move(feed)); // on_error } catch (const std::exception& e) { @@ -947,7 +981,7 @@ namespace lwsf { namespace internal bool wallet::connectToDaemon() { boost::unique_lock lock{data_.wallet->sync}; - if (data_.wallet->passed_login && data_.wallet->client.is_connected()) + if (data_.wallet->passed_login && (data_.wallet->feed || data_.wallet->client.is_connected())) return true; lock.unlock(); @@ -970,7 +1004,8 @@ namespace lwsf { namespace internal Monero::Wallet::ConnectionStatus wallet::connected() const { const boost::lock_guard lock{data_.wallet->sync}; - return data_.wallet->passed_login && data_.wallet->client.is_connected() ? + const bool feed_disabled = data_.wallet->feed_disabled; + return data_.wallet->passed_login && ((feed_disabled && data_.wallet->client.is_connected()) || (!feed_disabled && data_.wallet->feed)) ? ConnectionStatus_Connected : ConnectionStatus_Disconnected; } @@ -1309,12 +1344,22 @@ namespace lwsf { namespace internal unspent_map unspent{}; { wait_for( - [this] (auto&& callable) { backend::wallet::refresh(this->data_.wallet, false, std::move(callable)); } + [this] (auto&& f) { backend::wallet::refresh(this->data_.wallet, false, std::move(f)); } ); - const boost::lock_guard lock{data_.wallet->sync}; - fee_mask = data_.wallet->fee_mask; + boost::unique_lock lock{data_.wallet->sync}; + if (data_.wallet->need_fees()) // '/feed' doesn't fetch fees + { + lock.unlock(); + const std::error_code fee_fail = wait_for( + [this] (auto&& f) { backend::wallet::get_fees(this->data_.wallet, std::move(f)); } + ); + if (fee_fail) + throw std::system_error{fee_fail, "Tx construction"}; + lock.lock(); + } + fee_mask = data_.wallet->fee_mask; if (data_.wallet->per_byte_fee.empty() || !fee_mask || !data_.wallet->passed_login) throw std::system_error{data_.wallet->refresh_error, "Tx construction"}; diff --git a/src/wallet.h b/src/wallet.h index 05c6485..e114c76 100644 --- a/src/wallet.h +++ b/src/wallet.h @@ -70,7 +70,11 @@ namespace internal void operator()(std::error_code error) const; }; - bool thead_start(); + struct async_clear_error + { + std::shared_ptr self; + void operator()(std::error_code) const; + }; }; const net::wallet_io data_; diff --git a/tests/unit/backend.test.cpp b/tests/unit/backend.test.cpp index ec2d954..710ed7b 100644 --- a/tests/unit/backend.test.cpp +++ b/tests/unit/backend.test.cpp @@ -39,6 +39,7 @@ #include #include #include +#include #include #include #include @@ -56,9 +57,15 @@ #include "hex.h" // monero/contrib/include #include "rapidjson/document.h" // monero/external/rapidjson/include +#include "rapidjson/error/en.h" // monero/external/rapidjson/include #include "rapidjson/stringbuffer.h" // monero/external/rapidjson/incldue #include "rapidjson/writer.h" // monero/external/rapidjson/incldue #include "ringct/rctOps.h" // monero/src +#include "net/http.h" +#include "net/websocket.h" +#include "net/push.test.h" +#include "wire/adapted/crypto.h" +#include "wire/msgpack.h" namespace rapidjson { @@ -92,6 +99,73 @@ namespace return out; } + void to_msgpack(wire::msgpack_writer& dest, const rapidjson::Value& src) + { + switch (src.GetType()) + { + default: + case rapidjson::kNullType: + throw std::logic_error{"unexpected json value - " + std::to_string(src.GetType())}; + case rapidjson::kFalseType: + dest.boolean(false); + break; + case rapidjson::kTrueType: + dest.boolean(true); + break; + case rapidjson::kObjectType: + dest.start_object(src.MemberCount()); + { + const auto end = src.MemberEnd(); + for (auto cur = src.MemberBegin(); cur != end; ++cur) + { + dest.key({cur->name.GetString(), cur->name.GetStringLength()}); + to_msgpack(dest, cur->value); + } + } + dest.end_object(); + break; + case rapidjson::kArrayType: + dest.start_array(src.Size()); + { + for (std::size_t i = 0; i < src.Size(); ++i) + to_msgpack(dest, src[i]); + } + dest.end_array(); + break; + case rapidjson::kStringType: + { + std::string binary; + const boost::string_ref str{src.GetString(), src.GetStringLength()}; + if (epee::from_hex::to_string(binary, str)) + dest.binary(epee::to_byte_span(epee::to_span(binary))); + else + dest.string(str); + } + break; + case rapidjson::kNumberType: + if (src.IsUint()) dest.unsigned_integer(src.GetUint()); + else if (src.IsUint64()) dest.unsigned_integer(std::uintmax_t(src.GetUint64())); + else if (src.IsInt()) dest.integer(src.GetInt()); + else if (src.IsInt64()) dest.integer(std::intmax_t(src.GetInt64())); + else throw std::logic_error{"unexpected json number"}; + break; + } + } + + std::string to_msgpack(const std::string& src) + { + rapidjson::Document ir; + ir.Parse(src.c_str()); + if (ir.HasParseError()) + throw std::runtime_error{GetParseError_En(ir.GetParseError())}; + + wire::msgpack_slice_writer dest{}; + to_msgpack(dest, ir); + + const auto bytes = dest.take_sink(); + return std::string{reinterpret_cast(bytes.data()), bytes.size()}; + } + struct empty_slice { operator epee::byte_slice() const { return {}; } @@ -102,21 +176,49 @@ namespace operator lwsf::internal::rpc::get_random_outs_request() const { return {}; } }; + struct login + { + struct details + { + std::string address; + crypto::secret_key view_key; + }; + + details account; + }; + + void read_bytes(wire::reader& src, login::details& self) + { + wire::object(src, WIRE_FIELD(address), WIRE_FIELD(view_key)); + } + + void read_bytes(wire::reader& src, login& self) + { + wire::object(src, WIRE_FIELD(account)); + } + struct response_loop : boost::asio::coroutine { struct frame { lest::env& lest_env; - boost::asio::ip::tcp::socket server; + lwsf_test::net::push::socket server; boost::beast::flat_buffer buffer; boost::optional request; boost::beast::http::response response; std::deque> requests; std::deque> responses; + std::shared_ptr ws; explicit frame(lest::env& lest_env, boost::asio::io_context& io) - : lest_env(lest_env), server(io), buffer(), request(), response(), requests(), responses() + : lest_env(lest_env), server(io), buffer(), request(), response(), requests(), responses(), ws() {} + + ~frame() + { + if (ws) + lwsf_test::net::push::async_close(std::move(ws)); + } }; std::shared_ptr self; @@ -155,17 +257,33 @@ namespace { LWSF_VERIFY(!self->responses.empty()); - const auto string = boost::get(std::addressof(self->responses.front())); - if (string) - self->response = {boost::beast::http::status::ok, 11, *string}; - else - self->response = {boost::get(self->responses.front()), 11, ""}; + boost::variant response; + response.swap(self->responses.front()); + self->responses.pop_front(); + { + const auto string = boost::get(std::addressof(response)); + if (!string) + { + const auto status = boost::get(response); + if (status == boost::beast::http::status::switching_protocols) + { + auto self_ = self; + lwsf_test::net::push::async_handshake(std::move(self->server), *self->request, [self_] (auto ws) { + self_->ws = ws.value(); + }); + return; + } + else + self->response = {status, 11, std::string{}}; + } + else + self->response = {boost::beast::http::status::ok, 11, *string}; + } } self->response.keep_alive(true); self->response.prepare_payload(); BOOST_ASIO_CORO_YIELD boost::beast::http::async_write(self->server, self->response, *this); - self->responses.pop_front(); } } } @@ -250,7 +368,9 @@ namespace BOOST_ASIO_CORO_YIELD boost::beast::http::async_read(self->server, self->buffer, *self->request, *this); LWSF_VERIFY(self->request->keep_alive()); - if (self->dist(self->eng)) + if (self->request->target() == "/feed") + self->response = {boost::beast::http::status::not_found, 11, ""}; + else if (self->dist(self->eng)) self->response = {boost::beast::http::status::ok, 11, get_response(self->request->target())}; else self->response = {boost::beast::http::status::internal_server_error, 11, ""}; @@ -264,7 +384,7 @@ namespace }; } -LWS_CASE("backend::acount") +LWS_CASE("backend::account") { using subaddrs = lwsf::internal::rpc::subaddrs; @@ -307,7 +427,7 @@ LWS_CASE("backend::acount") } } -LWS_CASE("backend::wallet") +LWS_CASE("backend::wallet (non-websocket)") { using bwallet = lwsf::internal::backend::wallet; using tcp = boost::asio::ip::tcp; @@ -327,9 +447,10 @@ LWS_CASE("backend::wallet") SECTION("uninitialized http client") { using namespace std::placeholders; - const std::array)>, std::string>, 5> funcs1 + const std::array)>, std::string>, 6> funcs1 {{ {std::bind(bwallet::refresh, wallet, true, _1), "refresh"}, + {std::bind(bwallet::get_fees, wallet, _1), "get_fees"}, {std::bind(bwallet::register_subaccount, wallet, 0, _1), "register_subaccount"}, {std::bind(bwallet::register_subaddress, wallet, 0, 0, _1), "register_subaddress"}, {std::bind(bwallet::set_lookahead, wallet, 0, 0, _1), "set_lookahead"}, @@ -338,13 +459,15 @@ LWS_CASE("backend::wallet") for (const auto& func : funcs1) { + const lest::ctx ctx{lest_env, func.second}; + std::error_code actual{}; func.first([&] (auto result) { actual = result; }); io.restart(); io.run(); - SECTION(func.second) - EXPECT(actual == common_error::kInvalidArgument); + + EXPECT(actual == common_error::kInvalidArgument); } std::error_code actual = {}; @@ -542,10 +665,14 @@ LWS_CASE("backend::wallet") frame->requests = { {"/login", ""}, {"/upsert_subaddrs", ""}, + {"/feed", ""}, {"/get_address_txs", ""} }; frame->responses = { - R"({"new_address": true})", "", boost::beast::http::status::forbidden + R"({"new_address": true})", + "", + boost::beast::http::status::not_found, + boost::beast::http::status::forbidden }; bool done2 = false; @@ -570,6 +697,7 @@ LWS_CASE("backend::wallet") frame->requests = { {"/login", ""}, {"/upsert_subaddrs", ""}, + {"/feed", ""}, {"/get_address_txs", ""}, {"/get_unspent_outs", ""}, {"/get_version", ""} @@ -577,6 +705,7 @@ LWS_CASE("backend::wallet") frame->responses = { R"({"new_address": true})", "", + boost::beast::http::status::not_found, R"({ "total_received": "5000", "scanned_height": 150, @@ -664,11 +793,13 @@ LWS_CASE("backend::wallet") { frame->requests = { {"/login", ""}, + {"/feed", ""}, {"/get_address_txs", ""}, {"/get_unspent_outs", ""}, }; frame->responses = { R"({"new_address": true, "lookahead": {"maj_i": 50, "min_i": 200}})", + boost::beast::http::status::not_found, R"({ "total_received": "4000", "scanned_height": 175, @@ -795,6 +926,7 @@ LWS_CASE("backend::wallet") frame->requests = { {"/login", ""}, {"/upsert_subaddrs", upsert_body}, + {"/feed", ""}, {"/get_address_txs", ""}, {"/get_unspent_outs", ""}, {"/get_version", ""}, @@ -803,6 +935,7 @@ LWS_CASE("backend::wallet") frame->responses = { R"({"new_address": true, "lookahead": {"maj_i": 50, "min_i": 200}})", "", + boost::beast::http::status::not_found, R"({ "total_received": "4000", "scanned_height": 175, @@ -944,6 +1077,7 @@ LWS_CASE("backend::wallet") frame->requests = { {"/login", ""}, {"/upsert_subaddrs", upsert_body}, + {"/feed", ""}, {"/get_address_txs", ""}, {"/get_unspent_outs", ""}, {"/get_version", ""}, @@ -953,6 +1087,7 @@ LWS_CASE("backend::wallet") frame->responses = { R"({"new_address": true, "lookahead": {"maj_i": 50, "min_i": 200}})", "", + boost::beast::http::status::not_found, R"({ "total_received": "4000", "scanned_height": 175, @@ -1097,6 +1232,7 @@ LWS_CASE("backend::wallet") frame->requests = { {"/login", ""}, {"/upsert_subaddrs", upsert_body}, + {"/feed", ""}, {"/get_address_txs", ""}, {"/get_unspent_outs", ""}, {"/import_wallet_request", import_body} @@ -1104,6 +1240,7 @@ LWS_CASE("backend::wallet") frame->responses = { R"({"new_address": true, "lookahead": {"maj_i": 50, "min_i": 200}})", "", + boost::beast::http::status::not_found, R"({ "total_received": "4000", "scanned_height": 175, @@ -1216,10 +1353,11 @@ LWS_CASE("backend::wallet") std::atomic count{0}; const auto completed = [&] (auto) { ++count; }; - const std::array, 8> rpcs + const std::array, 9> rpcs {{ std::bind(bwallet::login_is_new, wallet, completed), std::bind(bwallet::refresh, wallet, true, completed), + std::bind(bwallet::get_fees, wallet, completed), std::bind(bwallet::register_subaccount, wallet, 1, completed), std::bind(bwallet::register_subaddress, wallet, 1, 0, completed), std::bind(bwallet::set_lookahead, wallet, 1, 1, completed), @@ -1290,3 +1428,675 @@ LWS_CASE("backend::wallet") } } +LWS_CASE("backend::wallet /feed (websocket)") +{ + using bwallet = lwsf::internal::backend::wallet; + using tcp = boost::asio::ip::tcp; + const auto no_ssl = epee::net_utils::ssl_support_t::e_ssl_support_disabled; + + SETUP("base empty wallet and basic server") + { + boost::asio::io_context io; + const auto wallet = std::make_shared(io); + crypto::generate_keys(wallet->primary.view.pub, wallet->primary.view.sec); + crypto::generate_keys(wallet->primary.spend.pub, wallet->primary.spend.sec); + wallet->primary.address = wallet->get_spend_address({0, 0}); + + tcp::acceptor acceptor{io, tcp::endpoint(tcp::v4(), 0)}; + acceptor.listen(); + + const auto login_verify = [&] (std::string sub) + { + EXPECT(boost::string_view{sub}.starts_with("login:")); + sub.erase(0, std::strlen("login:")); + + login user{}; + EXPECT(wire::msgpack::from_bytes(epee::byte_slice{std::move(sub)}, user) == std::error_code{}); + EXPECT(user.account.address == wallet->get_spend_address({0, 0})); + EXPECT(user.account.view_key == wallet->primary.view.sec); + }; + + const std::string upsert_body = + R"({"address":")" + wallet->get_spend_address({0, 0}) + + R"(","view_key":")" + to_hex(unwrap(unwrap(wallet->primary.view.sec))) + + R"(","subaddrs":[{"key":1,"value":[[0,2]]}],"get_all":false})"; + + const auto frame = std::make_shared(lest_env, io); + const auto frame2 = std::make_shared(lest_env, io); + wallet->client.init(io, "127.0.0.1", "", acceptor.local_endpoint().port(), no_ssl); + + bool done = false; + std::size_t refreshed = 0; + std::error_code result{}; + std::error_code ws_result{}; + const auto record_response = [&] (auto val) { done = true; result = val; }; + bwallet::refresh(wallet, true, record_response); + + wallet->passed_login = false; + wallet->feed_disabled = false; + wallet->on_feed_fail = [&] (auto val) { ws_result = val; }; + wallet->on_feed_refresh = [&] (auto err) { EXPECT(!err); ++refreshed; }; + acceptor.async_accept(frame->server, response_loop{frame}); + + SECTION("feed is aborted during upgrade") + { + wallet->on_feed_fail = [&] (auto err) { + EXPECT(err == boost::system::error_code{boost::beast::websocket::error::upgrade_declined}); + acceptor.async_accept(frame2->server, response_loop{frame2}); + }; + + frame->requests = { + {"/login", ""}, + {"/feed", ""}, + {"/feed", ""} + }; + frame->responses = { + R"({"new_address": true, "lookahead": {"maj_i": 50, "min_i": 200}})", + boost::beast::http::status::not_implemented, + boost::beast::http::status::not_found + }; + + frame2->requests = {{"/get_address_txs", ""}}; + frame2->responses = {boost::beast::http::status::forbidden}; + + while (!done) + { + io.restart(); + io.run_one(); + } + + EXPECT(refreshed == 0); + EXPECT(!wallet->on_feed_fail); + EXPECT(!wallet->on_feed_refresh); + EXPECT(frame->responses.size() == 0); + EXPECT(frame2->responses.size() == 0); + EXPECT(result == lwsf::error::approval); + EXPECT(!wallet->passed_login); + EXPECT(wallet->feed_disabled); + EXPECT(wallet->refresh_error == lwsf::error::approval); + } + + SECTION("feed is aborted after upgrade") + { + wallet->on_feed_fail = [&] (auto err) { + EXPECT(err == lwsf::error::bad_feed_prefix); + acceptor.async_accept(frame2->server, response_loop{frame2}); + }; + + const std::string error =R"({"msg": "account not found", "code": 1})"; + + frame->requests = { + {"/login", ""}, + {"/feed", ""}, + {"/feed", ""} + }; + frame->responses = { + R"({"new_address": true, "lookahead": {"maj_i": 50, "min_i": 200}})", + boost::beast::http::status::not_implemented, + boost::beast::http::status::switching_protocols + }; + + while (!frame->ws) + { + io.restart(); + io.run_one(); + } + + EXPECT(refreshed == 0); + EXPECT(wallet->on_feed_fail); + EXPECT(wallet->on_feed_refresh); + EXPECT(frame->responses.size() == 0); + EXPECT(frame2->responses.size() == 0); + EXPECT(result == std::error_code{}); + EXPECT(wallet->passed_login); + EXPECT(!wallet->feed_disabled); + EXPECT(wallet->refresh_error == std::error_code{}); + EXPECT(wallet->lookahead_error == std::error_code{}); + EXPECT(wallet->subaddress_error == std::error_code{}); + EXPECT(wallet->import_error == std::error_code{}); + + wallet->on_feed_refresh = + [&] (auto err) { EXPECT(err == lwsf::error::bad_feed_prefix); ++refreshed; }; + + lwsf_test::net::push::async_read(frame->ws, [&] (auto user) { + login_verify(user.value()); + lwsf_test::net::push::async_pub( + frame->ws, "error:" + to_msgpack(error), [&] (auto err) { EXPECT(!err); } + ); + }); + + frame2->requests = {{"/get_address_txs", ""}}; + frame2->responses = {boost::beast::http::status::forbidden}; + + while (!done) + { + io.restart(); + io.run_one(); + } + + EXPECT(refreshed == 1); + EXPECT(!wallet->on_feed_fail); + EXPECT(!wallet->on_feed_refresh); + EXPECT(frame->responses.size() == 0); + EXPECT(frame2->responses.size() == 0); + EXPECT(result == lwsf::error::approval); + EXPECT(!wallet->passed_login); + EXPECT(wallet->feed_disabled); + EXPECT(wallet->refresh_error == lwsf::error::approval); + } + + SECTION("feed sync (no initial txes)") + { + rct::key ringct{}; + lwsf::internal::backend::keypair tx_key{}; + lwsf::internal::backend::keypair output_key{}; + crypto::generate_keys(tx_key.pub, tx_key.sec); + const crypto::hash tx_hash = crypto::rand(); + const crypto::hash prefix_hash = crypto::rand(); + { + crypto::secret_key scalar{}; + crypto::key_derivation derived{}; + EXPECT(crypto::generate_key_derivation(tx_key.pub, wallet->primary.view.sec, derived)); + EXPECT(crypto::derive_public_key(derived, 3, wallet->primary.spend.pub, output_key.pub)); + crypto::derive_secret_key(derived, 3, wallet->primary.spend.sec, output_key.sec); + + crypto::derivation_to_scalar(derived, 3, scalar); + + rct::ecdhTuple commitment{}; + rct::ecdhDecode(commitment, rct::sk2rct(scalar), true); + ringct = commitment.mask; + } + + const std::string tx_sync = + R"({ + "scanned_block_height": 950, + "start_height": 900, + "blockchain_height": 1000, + "lookahead": {"maj_i": 50, "min_i": 200} + })"; + + const std::string blocks = + R"({ + "scan_start": 950, + "scan_end": 975, + "blockchain_height": 1001, + "transactions": [{ + "hash": ")" + to_hex(tx_hash) + R"(", + "prefix_hash": ")" + to_hex(prefix_hash) + R"(", + "timestamp": 9000, + "fee": 105, + "unlock_time": 5000, + "height": 961, + "mixin": 16, + "receives": [{ + "amount": 2566, + "public_key": ")" + to_hex(output_key.pub) + R"(", + "index": 3, + "id": {"legacy": {"amount": 0, "index": 1000}}, + "tx_pub_key": ")" + to_hex(tx_key.pub) + R"(", + "rct": ")" + to_hex(ringct) + R"(" + }] + }] + })"; + + const std::string warning = R"({"msg": "e", "height": 960, "code": 4})"; + + frame->requests = { + {"/login", ""}, + {"/feed", ""}, + {"/feed", ""} + }; + frame->responses = { + R"({"new_address": true, "lookahead": {"maj_i": 50, "min_i": 200}})", + boost::beast::http::status::not_implemented, + boost::beast::http::status::switching_protocols + }; + + while (!frame->ws) + { + io.restart(); + io.run_one(); + } + + EXPECT(!done); + EXPECT(refreshed == 0); + EXPECT(!wallet->feed_disabled); + EXPECT(wallet->on_feed_fail); + EXPECT(wallet->on_feed_refresh); + EXPECT(frame->responses.size() == 0); + EXPECT(result == std::error_code{}); + EXPECT(wallet->refresh_error == std::error_code{}); + EXPECT(wallet->lookahead_error == std::error_code{}); + EXPECT(wallet->subaddress_error == std::error_code{}); + EXPECT(wallet->import_error == std::error_code{}); + EXPECT(wallet->passed_login); + + lwsf_test::net::push::async_read(frame->ws, [&] (auto user) { + login_verify(user.value()); + lwsf_test::net::push::async_pub( + frame->ws, "tx_sync:" + to_msgpack(tx_sync), [&] (auto err) { EXPECT(!err); } + ); + }); + + while (!done) + { + io.restart(); + io.run_one(); + } + + EXPECT(refreshed == 1); + EXPECT(result == std::error_code{}); + EXPECT(ws_result == std::error_code{}); + EXPECT(wallet->on_feed_fail); + EXPECT(wallet->on_feed_refresh); + EXPECT(wallet->refresh_error == std::error_code{}); + EXPECT(wallet->lookahead_error == std::error_code{}); + EXPECT(wallet->subaddress_error == std::error_code{}); + EXPECT(wallet->import_error == std::error_code{}); + EXPECT(wallet->primary.restore_height == 900); + EXPECT(wallet->primary.scan_height == 950); + EXPECT(wallet->primary.requested_start == 900); + EXPECT(wallet->blockchain_height == 1000); + EXPECT(wallet->primary.txes.empty()); + + lwsf_test::net::push::async_pub( + frame->ws, "blocks:" + to_msgpack(blocks), [&] (auto err) { EXPECT(!err); } + ); + + while (refreshed <= 1) + { + io.restart(); + io.run_one(); + } + + EXPECT(refreshed == 2); + EXPECT(result == std::error_code{}); + EXPECT(ws_result == std::error_code{}); + EXPECT(!wallet->feed_disabled); + EXPECT(wallet->on_feed_fail); + EXPECT(wallet->on_feed_refresh); + EXPECT(wallet->refresh_error == std::error_code{}); + EXPECT(wallet->lookahead_error == std::error_code{}); + EXPECT(wallet->subaddress_error == std::error_code{}); + EXPECT(wallet->import_error == std::error_code{}); + EXPECT(wallet->primary.restore_height == 900); + EXPECT(wallet->primary.scan_height == 975); + EXPECT(wallet->primary.requested_start == 900); + EXPECT(wallet->blockchain_height == 1001); + EXPECT(wallet->primary.txes.size() == 1); + EXPECT(wallet->primary.txes.count(tx_hash) == 1); + const auto tx = wallet->primary.txes.at(tx_hash); + EXPECT(tx != nullptr); + EXPECT(tx->spends.empty()); + EXPECT(tx->receives.size() == 1); + EXPECT(tx->receives.count(output_key.pub) == 1); + const auto& in = tx->receives.at(output_key.pub); + EXPECT(in.global_index == 1000); + EXPECT(in.amount == 2566); + EXPECT(in.recipient == lwsf::internal::rpc::address_meta{}); + EXPECT(in.index == 3); + EXPECT(bool(in.rct_mask)); + EXPECT(in.rct_mask.value_or(rct::key{}) == ringct); + EXPECT(in.tx_pub == tx_key.pub); + EXPECT(tx->transfers.empty()); + EXPECT(tx->description.empty()); + EXPECT(bool(tx->timestamp)); + EXPECT(tx->timestamp.value_or(std::chrono::system_clock::time_point{}) == std::chrono::system_clock::time_point{std::chrono::seconds{9000}}); + EXPECT(bool(tx->height)); + EXPECT(tx->height.value_or(0) == 961); + EXPECT(tx->amount == 2566); + EXPECT(tx->fee == 105); + EXPECT(tx->unlock_time == 5000); + EXPECT(tx->direction == Monero::TransactionInfo::Direction_In); + EXPECT(std::get_if(std::addressof(tx->payment_id))); + EXPECT(tx->id == tx_hash); + EXPECT(tx->prefix == prefix_hash); + EXPECT(tx->coinbase == false); + EXPECT(tx->failed == false); + + lwsf_test::net::push::async_pub( + frame->ws, "warning:" + to_msgpack(warning), [&] (auto err) { EXPECT(!err); } + ); + + while (refreshed <= 2) + { + io.restart(); + io.run_one(); + } + + EXPECT(refreshed == 3); + EXPECT(result == std::error_code{}); + EXPECT(ws_result == std::error_code{}); + EXPECT(!wallet->feed_disabled); + EXPECT(wallet->on_feed_fail); + EXPECT(wallet->on_feed_refresh); + EXPECT(wallet->refresh_error == std::error_code{}); + EXPECT(wallet->lookahead_error == std::error_code{}); + EXPECT(wallet->subaddress_error == std::error_code{}); + EXPECT(wallet->import_error == std::error_code{}); + EXPECT(wallet->primary.restore_height == 900); + EXPECT(wallet->primary.scan_height == 960); + EXPECT(wallet->primary.requested_start == 900); + EXPECT(wallet->blockchain_height == 960); + EXPECT(wallet->primary.txes.empty()); + } + + SECTION("feed sync (initial txes)") + { + rct::key ringct{}; + rct::key ringct2{}; + crypto::key_image image{}; + lwsf::internal::backend::keypair tx_key{}; + lwsf::internal::backend::keypair tx_key2{}; + lwsf::internal::backend::keypair output_key{}; + lwsf::internal::backend::keypair output_key2{}; + crypto::generate_keys(tx_key.pub, tx_key.sec); + crypto::generate_keys(tx_key2.pub, tx_key2.sec); + const crypto::hash8 pid = crypto::rand(); + const crypto::hash tx_hash = crypto::rand(); + const crypto::hash tx_hash2 = crypto::rand(); + const crypto::hash prefix_hash = crypto::rand(); + const crypto::hash prefix_hash2 = crypto::rand(); + { + crypto::secret_key scalar{}; + crypto::key_derivation derived{}; + EXPECT(crypto::generate_key_derivation(tx_key.pub, wallet->primary.view.sec, derived)); + EXPECT(crypto::derive_public_key(derived, 0, wallet->primary.spend.pub, output_key.pub)); + crypto::derive_secret_key(derived, 0, wallet->primary.spend.sec, output_key.sec); + crypto::generate_key_image(output_key.pub, output_key.sec, image); + + crypto::derivation_to_scalar(derived, 0, scalar); + + rct::ecdhTuple commitment{}; + rct::ecdhDecode(commitment, rct::sk2rct(scalar), true); + ringct = commitment.mask; + } + { + crypto::secret_key scalar{}; + crypto::key_derivation derived{}; + EXPECT(crypto::generate_key_derivation(tx_key2.pub, wallet->primary.view.sec, derived)); + EXPECT(crypto::derive_public_key(derived, 3, wallet->primary.spend.pub, output_key2.pub)); + crypto::derive_secret_key(derived, 3, wallet->primary.spend.sec, output_key2.sec); + + crypto::derivation_to_scalar(derived, 3, scalar); + + rct::ecdhTuple commitment{}; + rct::ecdhDecode(commitment, rct::sk2rct(scalar), true); + ringct2 = commitment.mask; + } + + const std::string tx_sync = + R"({ + "scanned_block_height": 1950, + "start_height": 900, + "blockchain_height": 11000, + "lookahead": {"maj_i": 50, "min_i": 200}, + "transactions": [ + { + "hash": ")" + to_hex(tx_hash) + R"(", + "prefix_hash": ")" + to_hex(prefix_hash) + R"(", + "timestamp": 7000, + "fee": 103, + "unlock_time": 4000, + "height": 1955, + "mixin": 16, + "receives": [{ + "amount": 1366, + "public_key": ")" + to_hex(output_key.pub) + R"(", + "index": 0, + "id": {"legacy": {"amount": 0, "index": 1000}}, + "tx_pub_key": ")" + to_hex(tx_key.pub) + R"(", + "rct": ")" + to_hex(ringct) + R"(" + }] + },{ + "hash": ")" + to_hex(tx_hash2) + R"(", + "prefix_hash": ")" + to_hex(prefix_hash2) + R"(", + "timestamp": 9000, + "payment_id": ")" + to_hex(pid) + R"(", + "fee": 105, + "unlock_time": 5000, + "height": 1961, + "mixin": 16, + "receives": [ + { + "amount": 566, + "public_key": ")" + to_hex(output_key.pub) + R"(", + "index": 0, + "id": {"legacy": {"amount": 0, "index": 2000}}, + "tx_pub_key": ")" + to_hex(tx_key.pub) + R"(", + "rct": ")" + to_hex(ringct) + R"(" + },{ + "amount": 655, + "public_key": ")" + to_hex(output_key2.pub) + R"(", + "index": 3, + "id": {"legacy": {"amount": 0, "index": 2001}}, + "tx_pub_key": ")" + to_hex(tx_key2.pub) + R"(", + "rct": ")" + to_hex(ringct2) + R"(" + } + ], "spends": [{ + "id": {"legacy": {"amount": 0, "index": 1000}}, + "key_image": ")" + to_hex(image) + R"(" + }] + } + ] + })"; + + const std::string error = R"({"msg": "e", "code": 5})"; + + frame->requests = { + {"/login", ""}, + {"/feed", ""}, + {"/feed", ""} + }; + frame->responses = { + R"({"new_address": true, "lookahead": {"maj_i": 50, "min_i": 200}})", + boost::beast::http::status::not_implemented, + boost::beast::http::status::switching_protocols + }; + + while (!frame->ws) + { + io.restart(); + io.run_one(); + } + + EXPECT(!done); + EXPECT(refreshed == 0); + EXPECT(!wallet->feed_disabled); + EXPECT(wallet->on_feed_fail); + EXPECT(wallet->on_feed_refresh); + EXPECT(frame->responses.size() == 0); + EXPECT(result == std::error_code{}); + EXPECT(wallet->refresh_error == std::error_code{}); + EXPECT(wallet->lookahead_error == std::error_code{}); + EXPECT(wallet->subaddress_error == std::error_code{}); + EXPECT(wallet->import_error == std::error_code{}); + EXPECT(wallet->passed_login); + + lwsf_test::net::push::async_read(frame->ws, [&] (auto user) { + login_verify(user.value()); + lwsf_test::net::push::async_pub( + frame->ws, "tx_sync:" + to_msgpack(tx_sync), [&] (auto err) { EXPECT(!err); } + ); + }); + + while (!done) + { + io.restart(); + io.run_one(); + } + + EXPECT(refreshed == 1); + EXPECT(result == std::error_code{}); + EXPECT(ws_result == std::error_code{}); + EXPECT(wallet->on_feed_fail); + EXPECT(wallet->on_feed_refresh); + EXPECT(wallet->refresh_error == std::error_code{}); + EXPECT(wallet->lookahead_error == std::error_code{}); + EXPECT(wallet->subaddress_error == std::error_code{}); + EXPECT(wallet->import_error == std::error_code{}); + EXPECT(wallet->primary.restore_height == 900); + EXPECT(wallet->primary.scan_height == 1950); + EXPECT(wallet->primary.requested_start == 900); + EXPECT(wallet->blockchain_height == 11000); + EXPECT(wallet->primary.txes.size() == 2); + EXPECT(wallet->primary.txes.count(tx_hash) == 1); + { + const auto tx = wallet->primary.txes.at(tx_hash); + EXPECT(tx != nullptr); + EXPECT(tx->spends.empty()); + EXPECT(tx->receives.size() == 1); + EXPECT(tx->receives.count(output_key.pub) == 1); + const auto& in = tx->receives.at(output_key.pub); + EXPECT(in.global_index == 1000); + EXPECT(in.amount == 1366); + EXPECT(in.recipient == lwsf::internal::rpc::address_meta{}); + EXPECT(in.index == 0); + EXPECT(bool(in.rct_mask)); + EXPECT(in.rct_mask.value_or(rct::key{}) == ringct); + EXPECT(in.tx_pub == tx_key.pub); + EXPECT(tx->transfers.empty()); + EXPECT(tx->description.empty()); + EXPECT(bool(tx->timestamp)); + EXPECT(tx->timestamp.value_or(std::chrono::system_clock::time_point{}) == std::chrono::system_clock::time_point{std::chrono::seconds{7000}}); + EXPECT(bool(tx->height)); + EXPECT(tx->height.value_or(0) == 1955); + EXPECT(tx->amount == 1366); + EXPECT(tx->fee == 103); + EXPECT(tx->unlock_time == 4000); + EXPECT(tx->direction == Monero::TransactionInfo::Direction_In); + EXPECT(std::get_if(std::addressof(tx->payment_id))); + EXPECT(tx->id == tx_hash); + EXPECT(tx->prefix == prefix_hash); + EXPECT(tx->coinbase == false); + EXPECT(tx->failed == false); + } + { + const auto tx = wallet->primary.txes.at(tx_hash2); + EXPECT(tx != nullptr); + EXPECT(tx->spends.size() == 1); + EXPECT(tx->spends.count(image) == 1); + const auto& out = tx->spends.at(image); + EXPECT(out.amount == 1366); + EXPECT(out.sender == lwsf::internal::rpc::address_meta{}); + EXPECT(out.tx_pub == tx_key.pub); + EXPECT(out.output_pub == output_key.pub); + EXPECT(tx->receives.size() == 2); + EXPECT(tx->receives.count(output_key.pub) == 1); + EXPECT(tx->receives.count(output_key2.pub) == 1); + { + const auto& in = tx->receives.at(output_key.pub); + EXPECT(in.global_index == 2000); + EXPECT(in.amount == 566); + EXPECT(in.recipient == lwsf::internal::rpc::address_meta{}); + EXPECT(in.index == 0); + EXPECT(bool(in.rct_mask)); + EXPECT(in.rct_mask.value_or(rct::key{}) == ringct); + EXPECT(in.tx_pub == tx_key.pub); + } + { + const auto& in = tx->receives.at(output_key2.pub); + EXPECT(in.global_index == 2001); + EXPECT(in.amount == 655); + EXPECT(in.recipient == lwsf::internal::rpc::address_meta{}); + EXPECT(in.index == 3); + EXPECT(bool(in.rct_mask)); + EXPECT(in.rct_mask.value_or(rct::key{}) == ringct2); + EXPECT(in.tx_pub == tx_key2.pub); + } + EXPECT(tx->transfers.empty()); + EXPECT(tx->description.empty()); + EXPECT(bool(tx->timestamp)); + EXPECT(tx->timestamp.value_or(std::chrono::system_clock::time_point{}) == std::chrono::system_clock::time_point{std::chrono::seconds{9000}}); + EXPECT(bool(tx->height)); + EXPECT(tx->height.value_or(0) == 1961); + EXPECT(tx->amount == 40); + EXPECT(tx->fee == 105); + EXPECT(tx->unlock_time == 5000); + EXPECT(tx->direction == Monero::TransactionInfo::Direction_Out); + EXPECT(std::get_if(std::addressof(tx->payment_id))); + EXPECT(std::get(tx->payment_id) == pid); + EXPECT(tx->id == tx_hash2); + EXPECT(tx->prefix == prefix_hash2); + EXPECT(tx->coinbase == false); + EXPECT(tx->failed == false); + } + + wallet->on_feed_refresh = + [&] (auto err) { EXPECT(err == lwsf::error::feed); ++refreshed; }; + lwsf_test::net::push::async_pub( + frame->ws, "error:" + to_msgpack(error), [&] (auto err) { EXPECT(!err); } + ); + + while (refreshed <= 1) + { + io.restart(); + io.run_one(); + } + + EXPECT(refreshed == 2); + EXPECT(result == std::error_code{}); + EXPECT(ws_result == std::error_code{}); + EXPECT(wallet->on_feed_fail); + EXPECT(wallet->on_feed_refresh); + EXPECT(wallet->refresh_error == std::error_code{}); + EXPECT(wallet->lookahead_error == std::error_code{}); + EXPECT(wallet->subaddress_error == std::error_code{}); + EXPECT(wallet->import_error == std::error_code{}); + EXPECT(wallet->primary.restore_height == 900); + EXPECT(wallet->primary.scan_height == 1950); + EXPECT(wallet->primary.requested_start == 900); + EXPECT(wallet->blockchain_height == 11000); + EXPECT(wallet->primary.txes.size() == 2); + + acceptor.async_accept(frame2->server, response_loop{frame2}); + frame2->requests = {{"/feed", ""}, {"/feed", ""}}; + frame2->responses = { + boost::beast::http::status::not_implemented, + boost::beast::http::status::switching_protocols + }; + + while (!frame2->ws) + { + io.restart(); + io.run_one(); + } + + EXPECT(refreshed == 2); + EXPECT(result == std::error_code{}); + EXPECT(ws_result == std::error_code{}); + EXPECT(wallet->on_feed_fail); + EXPECT(wallet->on_feed_refresh); + EXPECT(wallet->refresh_error == std::error_code{}); + EXPECT(wallet->lookahead_error == std::error_code{}); + EXPECT(wallet->subaddress_error == std::error_code{}); + EXPECT(wallet->import_error == std::error_code{}); + EXPECT(wallet->primary.restore_height == 900); + EXPECT(wallet->primary.scan_height == 1950); + EXPECT(wallet->primary.requested_start == 900); + EXPECT(wallet->blockchain_height == 11000); + EXPECT(wallet->primary.txes.size() == 2); + } + + /* + * Websocket Cleanup + */ + + std::shared_ptr ws; + ws.swap(wallet->feed); + if (ws) + { + const std::size_t refreshed_count = refreshed; + wallet->on_feed_refresh = [&] (auto) { ++refreshed; }; + lwsf::internal::websocket::async_close(std::move(ws)); + lwsf_test::net::push::async_read(frame->ws, [] (auto) {}); + + while (refreshed_count == refreshed) + { + io.restart(); + io.run_one(); + } + } + } +} + diff --git a/tests/unit/net/CMakeLists.txt b/tests/unit/net/CMakeLists.txt index 0de1fc2..a05f164 100644 --- a/tests/unit/net/CMakeLists.txt +++ b/tests/unit/net/CMakeLists.txt @@ -26,6 +26,6 @@ # STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF # THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -add_library(lwsf-unit-net OBJECT http.test.cpp) +add_library(lwsf-unit-net OBJECT http.test.cpp push.test.cpp push.test.h) #target_include_directories(lwsf-unit-net PUBLIC ${CMAKE_CURRENT_SOURCE_DIR} "${CMAKE_SOURCE_DIR}/src") target_link_libraries(lwsf-unit-net PRIVATE lwsf-unit-framework lwsf-net monero::libraries) diff --git a/tests/unit/net/push.test.cpp b/tests/unit/net/push.test.cpp new file mode 100644 index 0000000..c93a577 --- /dev/null +++ b/tests/unit/net/push.test.cpp @@ -0,0 +1,149 @@ +// Copyright (c) 2026, The Monero Project +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without modification, are +// permitted provided that the following conditions are met: +// +// 1. Redistributions of source code must retain the above copyright notice, this list of +// conditions and the following disclaimer. +// +// 2. Redistributions in binary form must reproduce the above copyright notice, this list +// of conditions and the following disclaimer in the documentation and/or other +// materials provided with the distribution. +// +// 3. Neither the name of the copyright holder nor the names of its contributors may be +// used to endorse or promote products derived from this software without specific +// prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY +// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF +// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL +// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, +// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF +// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#include "push.test.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "error.h" + +namespace lwsf_test { namespace net { namespace push +{ + struct connection + { + boost::beast::websocket::stream ws_; + boost::asio::io_context::strand strand_; + std::deque>> msgs_; + boost::beast::flat_buffer buffer_; + + connection(socket&& sock) + : ws_(std::move(sock)), strand_(ws_.next_layer().get_executor().context()), msgs_(), buffer_() + { + ws_.binary(true); + } + + template + auto bind(F&& f) -> decltype(boost::asio::bind_executor(strand_, std::forward(f))) + { + return boost::asio::bind_executor(strand_, std::forward(f)); + } + }; + + namespace + { + struct on_read + { + std::shared_ptr self_; + std::function read_; + void operator()(const boost::system::error_code error, std::size_t) + { + LWSF_VERIFY(self_ && read_); + if (!error) + { + const auto buf = self_->buffer_.data(); + std::string out{reinterpret_cast(buf.data()), buf.size()}; + self_->buffer_.consume(buf.size()); + read_(std::move(out)); + } + else + read_(std::error_code{error}); + } + }; + + struct on_write + { + std::shared_ptr self_; + void operator()(const boost::system::error_code error, std::size_t = {}) + { + LWSF_VERIFY(self_ && !self_->msgs_.empty()); + + std::pair> msg; + msg.swap(self_->msgs_.front()); + self_->msgs_.pop_front(); + + if (!self_->msgs_.empty()) + start(std::move(self_)); + + if (msg.second) + msg.second(error); + } + + static void start(std::shared_ptr self) + { + LWSF_VERIFY(self && !self->msgs_.empty()); + const std::string& msg = self->msgs_.front().first; + self->ws_.async_write(boost::asio::buffer(msg.data(), msg.size()), self->bind(on_write{self})); + } + }; + } + + void async_handshake(socket&& sock, const request& req, std::function done) + { + auto self = std::make_shared(std::move(sock)); + self->ws_.async_accept(req, self->bind([self, done = std::move(done)] (auto error) { + if (error) + done(std::error_code{error}); + else + done(self); + })); + } + + void async_read(std::shared_ptr self, std::function read) + { + LWSF_VERIFY(self); + boost::asio::dispatch(self->strand_, [self, read = std::move(read)] () mutable { + self->ws_.async_read(self->buffer_, self->bind(on_read{self, std::move(read)})); + }); + } + + void async_pub(std::shared_ptr self, std::string msg, std::function sent) + { + LWSF_VERIFY(self); + boost::asio::dispatch(self->strand_, [self, msg = std::move(msg), sent = std::move(sent)] () mutable { + const bool empty = self->msgs_.empty(); + self->msgs_.push_back({std::move(msg), std::move(sent)}); + if (empty) + on_write::start(std::move(self)); + }); + } + + void async_close(std::shared_ptr self) + { + LWSF_VERIFY(self); + boost::asio::dispatch(self->strand_, [self] { + self->ws_.async_close(boost::beast::websocket::close_reason(), [] (auto) {}); + }); + } +}}} // lwsf_test // net // push diff --git a/tests/unit/net/push.test.h b/tests/unit/net/push.test.h new file mode 100644 index 0000000..611ba6e --- /dev/null +++ b/tests/unit/net/push.test.h @@ -0,0 +1,60 @@ +// Copyright (c) 2026, The Monero Project +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without modification, are +// permitted provided that the following conditions are met: +// +// 1. Redistributions of source code must retain the above copyright notice, this list of +// conditions and the following disclaimer. +// +// 2. Redistributions in binary form must reproduce the above copyright notice, this list +// of conditions and the following disclaimer in the documentation and/or other +// materials provided with the distribution. +// +// 3. Neither the name of the copyright holder nor the names of its contributors may be +// used to endorse or promote products derived from this software without specific +// prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY +// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF +// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL +// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, +// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF +// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "common/expect.h" + +namespace lwsf_test { namespace net { namespace push +{ + using socket = + boost::asio::basic_stream_socket< + boost::asio::ip::tcp, + boost::asio::io_context::executor_type + >; + using request = boost::beast::http::request; + + struct connection; + using ready = void(expect>); + using sub = void(expect); + using pub = void(std::error_code); + + void async_handshake(socket&& sock, const request& req, std::function done); + void async_read(std::shared_ptr self, std::function read); + void async_pub(std::shared_ptr self, std::string msg, std::function sent); + void async_close(std::shared_ptr self); + +}}} // lwsf_test // net // push