From 9eebe8e4d4533f51e0d2001d0f8ac0e22b95a8f3 Mon Sep 17 00:00:00 2001 From: Dario Date: Wed, 9 Sep 2026 11:13:55 +0200 Subject: [PATCH 1/7] sdk: replace async-wsocket with yawc for relay connections The relay transport now dials with yawc. The transport sink and stream carry yawc's `Frame` instead of `Message`, so incoming frames are matched on `OpCode` rather than on enum variants. Proxied connections go through yawc's own SOCKS5 client using `socks5h`, which leaves name resolution to the proxy and keeps `.onion` addresses reachable. yawc only offers rustls, so `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` are gone. `ring` and `aws_lc_rs` now select yawc's rustls backends, and `rustls-tls-webpki-roots` stays as a no-op because yawc always trusts the webpki roots. `nwc` and `nostr-connect` forward the reduced set. yawc's own `Stream` impl reports a read failure as the end of the stream, which would hide a broken connection behind a clean close, so the transport polls frames directly and keeps the two apart. The local relay's server side still needs async-wsocket, so that dependency stays for now, without its TLS features. --- Cargo.lock | 287 ++++++++------------------- nostr-sdk/Cargo.toml | 14 +- nostr-sdk/src/error.rs | 1 + nostr-sdk/src/relay/inner.rs | 79 ++++---- nostr-sdk/src/transport/websocket.rs | 281 ++++++++++++++++++++++---- nwc/Cargo.toml | 3 - signer/nostr-connect/Cargo.toml | 3 - 7 files changed, 371 insertions(+), 297 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 98f33b6ca..1d06891dd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -135,8 +135,6 @@ dependencies = [ "js-sys", "tokio", "tokio-happy-eyeballs", - "tokio-rustls", - "tokio-socks", "tokio-tungstenite", "url", "wasm-bindgen", @@ -360,7 +358,7 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" dependencies = [ - "nom", + "nom 7.1.3", ] [[package]] @@ -783,6 +781,7 @@ version = "1.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" dependencies = [ + "crc32fast", "miniz_oxide", "zlib-rs", ] @@ -817,21 +816,6 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" -[[package]] -name = "foreign-types" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6f339eb8adc052cd2ca78910fda869aefa38d22d5cb648e6485e4d3fc06f3b1" -dependencies = [ - "foreign-types-shared", -] - -[[package]] -name = "foreign-types-shared" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b" - [[package]] name = "form_urlencoded" version = "1.2.2" @@ -1200,9 +1184,9 @@ dependencies = [ [[package]] name = "hyper" -version = "1.9.0" +version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6299f016b246a94207e63da54dbe807655bf9e00044f73ded42c3ac5305fbcca" +checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43" dependencies = [ "atomic-waker", "bytes", @@ -1234,7 +1218,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots 1.0.6", + "webpki-roots", ] [[package]] @@ -1448,7 +1432,7 @@ dependencies = [ "jni-sys", "log", "simd_cesu8", - "thiserror 2.0.18", + "thiserror", "walkdir", "windows-link", ] @@ -1707,23 +1691,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "native-tls" -version = "0.2.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "465500e14ea162429d264d44189adc38b199b62b1c21eea9f69e4b73cb03bbf2" -dependencies = [ - "libc", - "log", - "openssl", - "openssl-probe", - "openssl-sys", - "schannel", - "security-framework", - "security-framework-sys", - "tempfile", -] - [[package]] name = "ndk-context" version = "0.1.1" @@ -1746,6 +1713,15 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "nom" +version = "8.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df9761775871bdef83bee530e60050f7e54b1105350d6884eb0fb4f46c2f9405" +dependencies = [ + "memchr", +] + [[package]] name = "nostr" version = "0.45.4" @@ -1959,6 +1935,7 @@ dependencies = [ "tracing", "tracing-subscriber", "universal-time", + "yawc", ] [[package]] @@ -1986,7 +1963,7 @@ dependencies = [ "futures", "libc", "libsodium-sys-stable", - "thiserror 2.0.18", + "thiserror", "tokio", "tracing", ] @@ -2062,59 +2039,6 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4f933a4265d5cdad61d19bbdfc972ea5726d56cd8d3d57b8f2d3c365dd42bee9" -[[package]] -name = "openssl" -version = "0.10.80" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a45fa2aa886c42762255da344f0a0d313e254066c46aad76f300c3d3da62d967" -dependencies = [ - "bitflags 2.11.0", - "cfg-if", - "foreign-types", - "libc", - "openssl-macros", - "openssl-sys", -] - -[[package]] -name = "openssl-macros" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "openssl-probe" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" - -[[package]] -name = "openssl-src" -version = "300.6.0+3.6.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a8e8cbfd3a4a8c8f089147fd7aaa33cf8c7450c4d09f8f80698a0cf093abeff4" -dependencies = [ - "cc", -] - -[[package]] -name = "openssl-sys" -version = "0.9.116" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f28a22dc7140cda5f096e5e7724a6962ca81a7f8bfd2979f9b18c11af56318c4" -dependencies = [ - "cc", - "libc", - "openssl-src", - "pkg-config", - "vcpkg", -] - [[package]] name = "page_size" version = "0.6.0" @@ -2206,6 +2130,26 @@ dependencies = [ "siphasher", ] +[[package]] +name = "pin-project" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -2286,7 +2230,7 @@ dependencies = [ "rustc-hash 2.1.1", "rustls", "socket2", - "thiserror 2.0.18", + "thiserror", "tokio", "tracing", "web-time", @@ -2307,7 +2251,7 @@ dependencies = [ "rustls", "rustls-pki-types", "slab", - "thiserror 2.0.18", + "thiserror", "tinyvec", "tracing", "web-time", @@ -2506,7 +2450,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", - "webpki-roots 1.0.6", + "webpki-roots", ] [[package]] @@ -2536,7 +2480,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a8a1f2315036ef6b1fbacd1972e8ee7688030b0a2121edfc2a6550febd41574d" dependencies = [ "hashbrown 0.16.1", - "thiserror 2.0.18", + "thiserror", ] [[package]] @@ -2607,6 +2551,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4" dependencies = [ "aws-lc-rs", + "log", "once_cell", "ring", "rustls-pki-types", @@ -2615,18 +2560,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "rustls-native-certs" -version = "0.8.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63" -dependencies = [ - "openssl-probe", - "rustls-pki-types", - "schannel", - "security-framework", -] - [[package]] name = "rustls-pki-types" version = "1.14.0" @@ -2680,15 +2613,6 @@ dependencies = [ "winapi-util", ] -[[package]] -name = "schannel" -version = "0.1.29" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91c1b7e4904c873ef0710c1f407dde2e6287de2bebc1bbbf7d430bb7cbffd939" -dependencies = [ - "windows-sys 0.61.2", -] - [[package]] name = "scopeguard" version = "1.2.0" @@ -2727,35 +2651,18 @@ dependencies = [ "cc", ] -[[package]] -name = "security-framework" -version = "3.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" -dependencies = [ - "bitflags 2.11.0", - "core-foundation", - "core-foundation-sys", - "libc", - "security-framework-sys", -] - -[[package]] -name = "security-framework-sys" -version = "2.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ce2691df843ecc5d231c0b14ece2acc3efb62c0a398c7e1d875f3983ce020e3" -dependencies = [ - "core-foundation-sys", - "libc", -] - [[package]] name = "semver" version = "1.0.27" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d767eb0aabc880b29956c35734170f26ed551a859dbd361d140cdbeca61ab1e2" +[[package]] +name = "send_wrapper" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd0b0ec5f1c1ca621c432a25813d8d60c88abe6d3e08a3eb9cf37d97a0fe3d73" + [[package]] name = "serde" version = "1.0.228" @@ -3017,33 +2924,13 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "thiserror" -version = "1.0.69" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" -dependencies = [ - "thiserror-impl 1.0.69", -] - [[package]] name = "thiserror" version = "2.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" dependencies = [ - "thiserror-impl 2.0.18", -] - -[[package]] -name = "thiserror-impl" -version = "1.0.69" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" -dependencies = [ - "proc-macro2", - "quote", - "syn", + "thiserror-impl", ] [[package]] @@ -3128,16 +3015,6 @@ dependencies = [ "syn", ] -[[package]] -name = "tokio-native-tls" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2" -dependencies = [ - "native-tls", - "tokio", -] - [[package]] name = "tokio-rustls" version = "0.26.4" @@ -3148,18 +3025,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "tokio-socks" -version = "0.5.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d4770b8024672c1101b3f6733eab95b18007dbe0847a8afe341fcf79e06043f" -dependencies = [ - "either", - "futures-util", - "thiserror 1.0.69", - "tokio", -] - [[package]] name = "tokio-stream" version = "0.1.18" @@ -3180,15 +3045,8 @@ checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" dependencies = [ "futures-util", "log", - "native-tls", - "rustls", - "rustls-native-certs", - "rustls-pki-types", "tokio", - "tokio-native-tls", - "tokio-rustls", "tungstenite", - "webpki-roots 0.26.11", ] [[package]] @@ -3327,12 +3185,9 @@ dependencies = [ "http", "httparse", "log", - "native-tls", "rand 0.9.2", - "rustls", - "rustls-pki-types", "sha1", - "thiserror 2.0.18", + "thiserror", "utf-8", ] @@ -3661,15 +3516,6 @@ dependencies = [ "web-sys", ] -[[package]] -name = "webpki-roots" -version = "0.26.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" -dependencies = [ - "webpki-roots 1.0.6", -] - [[package]] name = "webpki-roots" version = "1.0.6" @@ -4003,6 +3849,39 @@ dependencies = [ "rustix 1.1.4", ] +[[package]] +name = "yawc" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2c98759367e49a7a852c0256b94f18f8176740334c44401c0f688f37c6d7cb6" +dependencies = [ + "base64", + "bytes", + "flate2", + "futures", + "getrandom 0.2.17", + "http-body-util", + "hyper", + "hyper-util", + "js-sys", + "log", + "nom 8.0.0", + "percent-encoding", + "pin-project", + "rand 0.8.5", + "send_wrapper", + "sha1", + "thiserror", + "tokio", + "tokio-rustls", + "tokio-util", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "webpki-roots", +] + [[package]] name = "yoke" version = "0.8.1" diff --git a/nostr-sdk/Cargo.toml b/nostr-sdk/Cargo.toml index b837ad939..469d014d9 100644 --- a/nostr-sdk/Cargo.toml +++ b/nostr-sdk/Cargo.toml @@ -17,17 +17,16 @@ rustdoc-args = ["--cfg", "docsrs"] [features] default = ["ring", "rustls-tls-webpki-roots"] -aws_lc_rs = ["async-wsocket/aws_lc_rs"] -native-tls = ["async-wsocket/native-tls"] -native-tls-vendored = ["async-wsocket/native-tls-vendored"] -ring = ["async-wsocket/ring"] -rustls-tls-native-roots = ["async-wsocket/rustls-tls-native-roots"] -rustls-tls-webpki-roots = ["async-wsocket/rustls-tls-webpki-roots"] +aws_lc_rs = ["yawc/rustls-aws-lc-rs"] +ring = ["yawc/rustls-ring"] +# Kept for compatibility: yawc always trusts the webpki roots, so this selects nothing. +rustls-tls-webpki-roots = [] local-relay = ["dep:nostr-memory", "tokio/net"] [dependencies] async-utility.workspace = true -async-wsocket = { version = "0.17", features = ["socks"] } +# TODO: only the local relay's server side still needs this, drop it once that moves to yawc. +async-wsocket = { version = "0.17", default-features = false } faster-hex = { workspace = true, features = ["std"] } futures = "0.3" lru.workspace = true @@ -42,6 +41,7 @@ tokio = { workspace = true, features = ["macros", "sync"] } tokio-stream = { version = "0.1", features = ["sync"] } tracing = { workspace = true, features = ["std"] } universal-time = { workspace = true, features = ["std"] } +yawc = { version = "0.4", default-features = false } [dev-dependencies] base64 = { workspace = true, features = ["std"] } diff --git a/nostr-sdk/src/error.rs b/nostr-sdk/src/error.rs index 03bb4ba14..015180ff3 100644 --- a/nostr-sdk/src/error.rs +++ b/nostr-sdk/src/error.rs @@ -55,6 +55,7 @@ opaquerr::define_error! { negentropy::Error => ErrorKind::Protocol, #[cfg(any(feature = "local-relay", test))] async_wsocket::Error => ErrorKind::Other, + yawc::WebSocketError => ErrorKind::Transport, tokio::sync::oneshot::error::RecvError => ErrorKind::Other, tokio::sync::broadcast::error::RecvError => ErrorKind::Other, #[cfg(any(feature = "local-relay", test))] diff --git a/nostr-sdk/src/relay/inner.rs b/nostr-sdk/src/relay/inner.rs index 7a716b0bc..28404c1fa 100644 --- a/nostr-sdk/src/relay/inner.rs +++ b/nostr-sdk/src/relay/inner.rs @@ -1,13 +1,12 @@ use std::borrow::Cow; -use std::cmp; use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; +use std::{cmp, str}; use async_utility::{task, time}; -use async_wsocket::Message; use futures::{self, SinkExt, StreamExt}; use nostr::filter::MatchEventOptions; use nostr::message::MachineReadablePrefix; @@ -21,6 +20,7 @@ use rand::rngs::SysRng; use tokio::sync::mpsc::{self, Receiver, Sender}; use tokio::sync::{Mutex, MutexGuard, Notify, RwLock, RwLockWriteGuard, broadcast, oneshot, watch}; use universal_time::Instant; +use yawc::frame::{Frame, OpCode}; use super::capabilities::{AtomicRelayCapabilities, RelayCapabilities}; use super::constants::{ @@ -884,7 +884,7 @@ impl InnerRelay { tracing::debug!("Sending '{json}' to '{}' (size: {size} bytes)", self.url); // Compose WebSocket text messages - let msg: Message = Message::Text(json); + let msg: Frame = Frame::text(json); // Send WebSocket messages send_ws_msg(ws_tx, msg).await?; @@ -916,7 +916,7 @@ impl InnerRelay { ping.set_replied(false); // Compose ping message - let msg = Message::Ping(nonce.to_be_bytes().to_vec()); + let msg = Frame::ping(nonce.to_be_bytes().to_vec()); // Send WebSocket message send_ws_msg(ws_tx, msg).await?; @@ -944,47 +944,54 @@ impl InnerRelay { #[cfg(target_arch = "wasm32")] let _ping = ping; - while let Some(msg) = ws_rx.next().await { - match msg? { - Message::Text(json) => self.handle_relay_message(&json, &ingester_tx).await, - Message::Binary(_) => { + while let Some(frame) = ws_rx.next().await { + let frame: Frame = frame?; + + match frame.opcode() { + // The transport asks yawc to validate UTF-8, so an invalid payload closes the + // connection before it reaches here. Handle it anyway rather than panicking on + // `Frame::as_str`. + OpCode::Text => match str::from_utf8(frame.payload()) { + Ok(json) => self.handle_relay_message(json, &ingester_tx).await, + Err(..) => { + return Err(Error::protocol_msg("relay message is not valid UTF-8")); + } + }, + OpCode::Binary => { tracing::warn!(url = %self.url, "Binary messages aren't supported."); } #[cfg(not(target_arch = "wasm32"))] - Message::Pong(bytes) if self.opts.ping && self.state.transport.support_ping() => { - match bytes.try_into() { - Ok(nonce) => { - // Nonce from big-endian bytes - let nonce: u64 = u64::from_be_bytes(nonce); - - // Get last nonce - let last_nonce: u64 = ping.last_nonce(); - - // Check if last nonce not matches the received one - if last_nonce != nonce { - return Err(Error::pong_not_match(last_nonce, nonce)); - } + OpCode::Pong if self.opts.ping && self.state.transport.support_ping() => { + let Ok(nonce) = <[u8; 8]>::try_from(frame.payload().as_ref()) else { + return Err(Error::protocol_msg("can't parse pong")); + }; - // Set ping as replied - ping.set_replied(true); + // Nonce from big-endian bytes + let nonce: u64 = u64::from_be_bytes(nonce); - // Save latency - let sent_at = ping.sent_at().await; - self.stats.save_latency(sent_at.elapsed()); - } - Err(..) => { - return Err(Error::protocol_msg("can't parse pong")); - } + // Get last nonce + let last_nonce: u64 = ping.last_nonce(); + + // Check if last nonce not matches the received one + if last_nonce != nonce { + return Err(Error::pong_not_match(last_nonce, nonce)); } + + // Set ping as replied + ping.set_replied(true); + + // Save latency + let sent_at = ping.sent_at().await; + self.stats.save_latency(sent_at.elapsed()); } #[cfg(not(target_arch = "wasm32"))] - Message::Close(None) => break, - #[cfg(not(target_arch = "wasm32"))] - Message::Close(Some(frame)) => { - tracing::info!(code = %frame.code, reason = %frame.reason, "Connection closed by peer."); + OpCode::Close => { + if let Some(code) = frame.close_code() { + let reason: &str = frame.close_reason().ok().flatten().unwrap_or_default(); + tracing::info!(?code, reason, "Connection closed by peer."); + } break; } - #[cfg(not(target_arch = "wasm32"))] _ => {} } } @@ -1857,7 +1864,7 @@ impl InnerRelay { } /// Send a WebSocket message with timeout set to [WEBSOCKET_TX_TIMEOUT]. -async fn send_ws_msg(tx: &mut WebSocketSink, msg: Message) -> Result<(), Error> { +async fn send_ws_msg(tx: &mut WebSocketSink, msg: Frame) -> Result<(), Error> { match time::timeout(Some(WEBSOCKET_TX_TIMEOUT), tx.send(msg)).await { Some(res) => Ok(res?), None => Err(Error::timeout()), diff --git a/nostr-sdk/src/transport/websocket.rs b/nostr-sdk/src/transport/websocket.rs index 47358d53b..63a5ffbbe 100644 --- a/nostr-sdk/src/transport/websocket.rs +++ b/nostr-sdk/src/transport/websocket.rs @@ -10,12 +10,15 @@ use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; -use async_wsocket::{ConnectionMode, Message, WebSocket}; -#[cfg(not(target_arch = "wasm32"))] -use async_wsocket::{HeaderMap, HeaderValue}; use futures::stream::SplitSink; use futures::{Sink, SinkExt, Stream, StreamExt, TryStreamExt}; use nostr::types::Url; +#[cfg(target_arch = "wasm32")] +use yawc::WebSocket; +use yawc::WebSocketError; +use yawc::frame::Frame; +#[cfg(not(target_arch = "wasm32"))] +use yawc::{HttpRequest, Options, Proxy, TcpWebSocket, WebSocket}; use crate::error::Error; use crate::future::BoxedFuture; @@ -23,10 +26,18 @@ use crate::future::BoxedFuture; #[cfg(not(target_arch = "wasm32"))] const USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION")); +/// Largest single frame payload accepted from a relay. +#[cfg(not(target_arch = "wasm32"))] +const MAX_PAYLOAD_READ: usize = 16 * 1024 * 1024; + +/// Largest message accepted from a relay once its fragments are reassembled. +#[cfg(not(target_arch = "wasm32"))] +const MAX_READ_BUFFER: usize = 64 * 1024 * 1024; + /// WebSocket transport sink -pub type WebSocketSink = Pin + Send>>; +pub type WebSocketSink = Pin + Send>>; /// WebSocket transport stream -pub type WebSocketStream = Pin> + Send>>; +pub type WebSocketStream = Pin> + Send>>; #[doc(hidden)] pub trait IntoWebSocketTransport { @@ -85,41 +96,144 @@ impl WebSocketTransport for DefaultWebsocketTransport { proxy: Option, ) -> BoxedFuture<'a, Result<(WebSocketSink, WebSocketStream), Error>> { Box::pin(async move { - let mode: ConnectionMode = match proxy { - #[cfg(not(target_arch = "wasm32"))] - Some(proxy) => ConnectionMode::Proxy(proxy), - #[cfg(target_arch = "wasm32")] - Some(_) => ConnectionMode::Direct, - None => ConnectionMode::Direct, - }; - #[cfg(not(target_arch = "wasm32"))] - let connection = { - let mut headers = HeaderMap::new(); - headers.insert("user-agent", HeaderValue::from_static(USER_AGENT)); - WebSocket::connect_with_headers(url, &mode, headers) - }; + { + let socket: TcpWebSocket = connect_native(url, proxy).await?; + Ok(split(Reporting::new(socket))) + } + #[cfg(target_arch = "wasm32")] - let connection = WebSocket::connect(url, &mode); + { + // The browser dials on our behalf, so a proxy can't be applied here. + let _ = proxy; + let socket: WebSocket = WebSocket::connect(url.clone()) + .await + .map_err(Error::transport)?; + Ok(split(socket)) + } + }) + } +} + +#[cfg(not(target_arch = "wasm32"))] +async fn connect_native(url: &Url, proxy: Option) -> Result { + let options: Options = Options::default() + .with_limits(MAX_PAYLOAD_READ, MAX_READ_BUFFER) + .with_utf8(); + + let request = HttpRequest::builder().header("user-agent", USER_AGENT); + + let mut builder = WebSocket::connect(url.clone()) + .with_options(options) + .with_request(request); + + if let Some(proxy) = proxy { + // `socks5h` leaves name resolution to the proxy, which is what makes `.onion` + // addresses resolvable at all. + let url: Url = Url::parse(&format!("socks5h://{proxy}")).map_err(Error::transport)?; + builder = builder.with_proxy(Proxy::socks5(url).map_err(Error::transport)?); + } + + builder.await.map_err(Error::transport) +} - let socket: WebSocket = connection.await.map_err(Error::transport)?; +/// Split a socket into the boxed sink and stream halves the transport hands back. +fn split(socket: T) -> (WebSocketSink, WebSocketStream) +where + T: Sink + + Stream> + + Send + + Unpin + + 'static, +{ + let (tx, rx) = socket.split(); - // Split sink and stream - let (tx, rx) = socket.split(); + // NOTE: don't use sink_map_err here, as it may cause panics! + // Issue: https://github.com/nostrdevkit/nostr/issues/984 + let sink: WebSocketSink = Box::pin(TransportSink(tx)) as WebSocketSink; + let stream: WebSocketStream = Box::pin(rx.map_err(Error::transport)) as WebSocketStream; - // NOTE: don't use sink_map_err here, as it may cause panics! - // Issue: https://github.com/nostrdevkit/nostr/issues/984 - let sink: WebSocketSink = Box::pin(TransportSink(tx)) as WebSocketSink; - let stream: WebSocketStream = Box::pin(rx.map_err(Error::transport)) as WebSocketStream; + (sink, stream) +} - Ok((sink, stream)) - }) +/// A socket that reports the read errors yawc's own [`Stream`] impl hides. +/// +/// `impl Stream for WebSocket` maps a failure to the end of the stream, which would make a +/// broken connection indistinguishable from one the relay closed cleanly. Polling the frame +/// directly keeps the two apart, at the cost of tracking the terminal state here. +#[cfg(not(target_arch = "wasm32"))] +struct Reporting { + socket: WebSocket, + /// Set once a read has failed, so the socket is not polled again afterwards. + failed: bool, +} + +#[cfg(not(target_arch = "wasm32"))] +impl Reporting { + fn new(socket: WebSocket) -> Self { + Self { + socket, + failed: false, + } } } -struct TransportSink(SplitSink); +#[cfg(not(target_arch = "wasm32"))] +impl Stream for Reporting +where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, +{ + type Item = Result; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = self.get_mut(); -impl Sink for TransportSink { + if this.failed { + return Poll::Ready(None); + } + + match futures::ready!(this.socket.poll_next_frame(cx)) { + Ok(frame) => Poll::Ready(Some(Ok(frame))), + Err(e) => { + this.failed = true; + Poll::Ready(Some(Err(e))) + } + } + } +} + +#[cfg(not(target_arch = "wasm32"))] +impl Sink for Reporting +where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, +{ + type Error = WebSocketError; + + fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.socket).poll_ready(cx) + } + + fn start_send(mut self: Pin<&mut Self>, item: Frame) -> Result<(), Self::Error> { + Pin::new(&mut self.socket).start_send(item) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.socket).poll_flush(cx) + } + + fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.socket).poll_close(cx) + } +} + +struct TransportSink(SplitSink) +where + T: Sink; + +impl Sink for TransportSink +where + T: Sink + Unpin, +{ type Error = Error; fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { @@ -128,7 +242,7 @@ impl Sink for TransportSink { .map_err(Error::transport) } - fn start_send(mut self: Pin<&mut Self>, item: Message) -> Result<(), Self::Error> { + fn start_send(mut self: Pin<&mut Self>, item: Frame) -> Result<(), Self::Error> { Pin::new(&mut self.0) .start_send_unpin(item) .map_err(Error::transport) @@ -149,11 +263,30 @@ impl Sink for TransportSink { #[cfg(all(test, not(target_arch = "wasm32")))] mod tests { - use tokio::io::AsyncReadExt; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; use super::*; + /// Read the HTTP request a client wrote, up to the blank line that ends its head. + async fn read_request(stream: &mut S) -> String + where + S: tokio::io::AsyncRead + Unpin, + { + let mut request = [0u8; 4096]; + let mut len = 0; + + while len < request.len() && !request[..len].ends_with(b"\r\n\r\n") { + let read = stream.read(&mut request[len..]).await.unwrap(); + if read == 0 { + break; + } + len += read; + } + + String::from_utf8(request[..len].to_vec()).unwrap() + } + #[tokio::test] async fn default_transport_sends_user_agent() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -161,18 +294,7 @@ mod tests { let server = tokio::spawn(async move { let (mut stream, _) = listener.accept().await.unwrap(); - let mut request = [0u8; 4096]; - let mut len = 0; - - while len < request.len() && !request[..len].ends_with(b"\r\n\r\n") { - let read = stream.read(&mut request[len..]).await.unwrap(); - if read == 0 { - break; - } - len += read; - } - - String::from_utf8(request[..len].to_vec()).unwrap() + read_request(&mut stream).await }); let url = Url::parse(&format!("ws://{address}")).unwrap(); @@ -186,4 +308,75 @@ mod tests { }); assert_eq!(user_agent, Some(USER_AGENT)); } + + /// Serve one SOCKS5 CONNECT, then read whatever the client tunnels through it. + /// + /// Nothing answers the WebSocket handshake, so the client's connect fails. What the test + /// cares about is the address the proxy was asked to reach and the bytes that followed. + async fn socks5_connect(listener: TcpListener) -> (String, String) { + let (mut stream, _) = listener.accept().await.unwrap(); + + // Greeting: version, then the authentication methods the client offers. + let mut greeting = [0u8; 2]; + stream.read_exact(&mut greeting).await.unwrap(); + assert_eq!(greeting[0], 0x05); + let mut methods = vec![0u8; greeting[1] as usize]; + stream.read_exact(&mut methods).await.unwrap(); + + // "No authentication required" + stream.write_all(&[0x05, 0x00]).await.unwrap(); + + // Request: version, command, reserved, address type. + let mut request = [0u8; 4]; + stream.read_exact(&mut request).await.unwrap(); + assert_eq!(request[0], 0x05); + assert_eq!(request[1], 0x01, "expected a CONNECT command"); + assert_eq!( + request[3], 0x03, + "expected a domain, left for the proxy to resolve" + ); + + let mut length = [0u8; 1]; + stream.read_exact(&mut length).await.unwrap(); + let mut domain = vec![0u8; length[0] as usize]; + stream.read_exact(&mut domain).await.unwrap(); + let mut port = [0u8; 2]; + stream.read_exact(&mut port).await.unwrap(); + + let target: String = format!( + "{}:{}", + String::from_utf8(domain).unwrap(), + u16::from_be_bytes(port) + ); + + // Success, with a bound address of 0.0.0.0:0. + stream + .write_all(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) + .await + .unwrap(); + + (target, read_request(&mut stream).await) + } + + #[tokio::test] + async fn default_transport_dials_through_socks5_proxy() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy = listener.local_addr().unwrap(); + + let server = tokio::spawn(socks5_connect(listener)); + + // A name that never resolves locally, so reaching it proves the proxy resolved it. + // This is what makes `.onion` addresses work. + let url = Url::parse("ws://relay.invalid:8080").unwrap(); + assert!( + DefaultWebsocketTransport + .connect(&url, Some(proxy)) + .await + .is_err() + ); + + let (target, request) = server.await.unwrap(); + assert_eq!(target, "relay.invalid:8080"); + assert!(request.starts_with("GET / HTTP/1.1"), "{request}"); + } } diff --git a/nwc/Cargo.toml b/nwc/Cargo.toml index 733a52c58..8e483c640 100644 --- a/nwc/Cargo.toml +++ b/nwc/Cargo.toml @@ -18,10 +18,7 @@ rustdoc-args = ["--cfg", "docsrs"] [features] default = ["ring", "rustls-tls-webpki-roots"] aws_lc_rs = ["nostr-sdk/aws_lc_rs"] -native-tls = ["nostr-sdk/native-tls"] -native-tls-vendored = ["nostr-sdk/native-tls-vendored"] ring = ["nostr-sdk/ring"] -rustls-tls-native-roots = ["nostr-sdk/rustls-tls-native-roots"] rustls-tls-webpki-roots = ["nostr-sdk/rustls-tls-webpki-roots"] [dependencies] diff --git a/signer/nostr-connect/Cargo.toml b/signer/nostr-connect/Cargo.toml index f4a9d8375..53da1c8ef 100644 --- a/signer/nostr-connect/Cargo.toml +++ b/signer/nostr-connect/Cargo.toml @@ -18,10 +18,7 @@ rustdoc-args = ["--cfg", "docsrs"] [features] default = ["ring", "rustls-tls-webpki-roots"] aws_lc_rs = ["nostr-sdk/aws_lc_rs"] -native-tls = ["nostr-sdk/native-tls"] -native-tls-vendored = ["nostr-sdk/native-tls-vendored"] ring = ["nostr-sdk/ring"] -rustls-tls-native-roots = ["nostr-sdk/rustls-tls-native-roots"] rustls-tls-webpki-roots = ["nostr-sdk/rustls-tls-webpki-roots"] [dependencies] From 71eacb3a8c857a9b892109f827b5e70e5806c70f Mon Sep 17 00:00:00 2001 From: Dario Date: Wed, 9 Sep 2026 11:21:29 +0200 Subject: [PATCH 2/7] sdk(local_relay): serve WebSocket connections with hyper and yawc yawc's server side is reached through an HTTP upgrade, so the listener now serves HTTP/1 with hyper and calls `WebSocket::upgrade_with_options`. The upgraded socket comes back over a channel, which keeps the handshake timeout and the connection permits where they were. `LocalRelay::take_connection` is unchanged: `WebSocket::from_stream` wraps a stream whose handshake someone else performed, so callers that upgrade the request themselves keep working. The session's frames are matched on `OpCode`, and the error-reporting socket wrapper is shared with the client transport so a failed read is still logged rather than read as a clean close. With this, async-wsocket is gone. --- Cargo.lock | 69 ------ nostr-sdk/Cargo.toml | 6 +- nostr-sdk/src/error.rs | 2 - nostr-sdk/src/local_relay/local/inner.rs | 298 ++++++++++++----------- nostr-sdk/src/local_relay/local/mod.rs | 254 +++++++------------ nostr-sdk/src/transport/websocket.rs | 4 +- 6 files changed, 260 insertions(+), 373 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1d06891dd..2aaea345b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -124,24 +124,6 @@ dependencies = [ "wasm-bindgen-futures", ] -[[package]] -name = "async-wsocket" -version = "0.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2c713e1f14c7b82e32ea159af1c6e2f070cfadbdf23fb2512acce9af0a26f1a2" -dependencies = [ - "futures", - "futures-util", - "js-sys", - "tokio", - "tokio-happy-eyeballs", - "tokio-tungstenite", - "url", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "atomic-destructor" version = "0.3.0" @@ -614,12 +596,6 @@ version = "0.3.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06d2e3287df1c007e74221c49ca10a95d557349e54b3a75dc2fb14712c751f04" -[[package]] -name = "data-encoding" -version = "2.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7a1e2f27636f116493b8b860f5546edb47c8d8f8ea73e1d2a20be88e28d1fea" - [[package]] name = "dialoguer" version = "0.12.0" @@ -1911,7 +1887,6 @@ name = "nostr-sdk" version = "0.45.2" dependencies = [ "async-utility", - "async-wsocket", "base64", "bitcoin_hashes 1.1.0", "faster-hex", @@ -2995,15 +2970,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "tokio-happy-eyeballs" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8564c32dfb6f4257f8bc6edfc178a34af97520e0b7b9815500c55eb3d092f29f" -dependencies = [ - "tokio", -] - [[package]] name = "tokio-macros" version = "2.6.1" @@ -3037,18 +3003,6 @@ dependencies = [ "tokio-util", ] -[[package]] -name = "tokio-tungstenite" -version = "0.28.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" -dependencies = [ - "futures-util", - "log", - "tokio", - "tungstenite", -] - [[package]] name = "tokio-util" version = "0.7.18" @@ -3174,23 +3128,6 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" -[[package]] -name = "tungstenite" -version = "0.28.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" -dependencies = [ - "bytes", - "data-encoding", - "http", - "httparse", - "log", - "rand 0.9.2", - "sha1", - "thiserror", - "utf-8", -] - [[package]] name = "typed-path" version = "0.12.3" @@ -3290,12 +3227,6 @@ dependencies = [ "serde_derive", ] -[[package]] -name = "utf-8" -version = "0.7.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" - [[package]] name = "utf8-zero" version = "0.8.1" diff --git a/nostr-sdk/Cargo.toml b/nostr-sdk/Cargo.toml index 469d014d9..e4fd5d2c1 100644 --- a/nostr-sdk/Cargo.toml +++ b/nostr-sdk/Cargo.toml @@ -21,14 +21,14 @@ aws_lc_rs = ["yawc/rustls-aws-lc-rs"] ring = ["yawc/rustls-ring"] # Kept for compatibility: yawc always trusts the webpki roots, so this selects nothing. rustls-tls-webpki-roots = [] -local-relay = ["dep:nostr-memory", "tokio/net"] +local-relay = ["dep:hyper", "dep:hyper-util", "dep:nostr-memory", "tokio/net"] [dependencies] async-utility.workspace = true -# TODO: only the local relay's server side still needs this, drop it once that moves to yawc. -async-wsocket = { version = "0.17", default-features = false } faster-hex = { workspace = true, features = ["std"] } futures = "0.3" +hyper = { version = "1.9", features = ["server", "http1"], optional = true } +hyper-util = { version = "0.1", features = ["tokio"], optional = true } lru.workspace = true negentropy = { workspace = true, features = ["std"] } nostr = { workspace = true, features = ["std", "rand", "os-rng"] } diff --git a/nostr-sdk/src/error.rs b/nostr-sdk/src/error.rs index 015180ff3..d60cfb2d3 100644 --- a/nostr-sdk/src/error.rs +++ b/nostr-sdk/src/error.rs @@ -53,8 +53,6 @@ opaquerr::define_error! { nostr_gossip::error::Error => ErrorKind::Gossip, faster_hex::Error => ErrorKind::Protocol, negentropy::Error => ErrorKind::Protocol, - #[cfg(any(feature = "local-relay", test))] - async_wsocket::Error => ErrorKind::Other, yawc::WebSocketError => ErrorKind::Transport, tokio::sync::oneshot::error::RecvError => ErrorKind::Other, tokio::sync::broadcast::error::RecvError => ErrorKind::Other, diff --git a/nostr-sdk/src/local_relay/local/inner.rs b/nostr-sdk/src/local_relay/local/inner.rs index a49036972..4b00d6746 100644 --- a/nostr-sdk/src/local_relay/local/inner.rs +++ b/nostr-sdk/src/local_relay/local/inner.rs @@ -12,8 +12,10 @@ use std::time::Duration; use async_utility::futures_util::stream::SplitSink; use async_utility::futures_util::{SinkExt, StreamExt}; -use async_wsocket::native; -use async_wsocket::native::{Message, Role, WebSocketConfig, WebSocketStream}; +use hyper::body::Incoming; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper_util::rt::TokioIo; use negentropy::{Id, Negentropy, NegentropyStorageVector}; use nostr::filter::{MatchEventOptions, SingleLetterTag}; use nostr::message::MachineReadablePrefix; @@ -21,7 +23,9 @@ use nostr::prelude::*; use nostr_memory::prelude::*; use tokio::io::{AsyncRead, AsyncWrite}; use tokio::net::TcpListener; -use tokio::sync::{Notify, OnceCell, OwnedSemaphorePermit, Semaphore, broadcast}; +use tokio::sync::{Notify, OnceCell, OwnedSemaphorePermit, Semaphore, broadcast, mpsc}; +use yawc::frame::{Frame, OpCode}; +use yawc::{Options, Role, WebSocket}; use super::super::builder::{ DEFAULT_MAX_PENDING_HANDSHAKES, LocalRelayBuilder, LocalRelayBuilderMode, @@ -33,8 +37,9 @@ use super::util; use crate::client::{Client, ClientNotification, Output, RelayUrlArg, SyncSummary}; use crate::error::{Error, ErrorKind}; use crate::relay::SyncOptions; +use crate::transport::websocket::Reporting; -type WsTx = SplitSink, Message>; +type WsTx = SplitSink, Frame>; #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum GiftWrapQueryAccess { @@ -289,11 +294,11 @@ impl InnerLocalRelay { tokio::time::sleep(unresponsive_connection).await; } - let ws_stream = - WebSocketStream::from_raw_socket(stream, Role::Server, Some(self.websocket_config())) - .await; + let socket = WebSocket::from_stream(stream, Role::Server, self.websocket_options()) + .map_err(Error::transport)?; - self.handle_websocket(ws_stream, addr, permit).await?; + self.handle_websocket(Reporting::new(socket), addr, permit) + .await?; Ok(()) } @@ -306,22 +311,53 @@ impl InnerLocalRelay { permit: OwnedSemaphorePermit, ) -> Result<(), Error> where - S: AsyncRead + AsyncWrite + Unpin, + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { if let Some(unresponsive_connection) = self.test.unresponsive_connection { tokio::time::sleep(unresponsive_connection).await; } + // The upgraded socket only exists once the response has been written, so the service + // hands it back over a channel instead of returning it. + let (upgraded_tx, mut upgraded_rx) = mpsc::channel(1); + let options: Options = self.websocket_options(); + + let service = service_fn(move |mut request: hyper::Request| { + let upgraded_tx = upgraded_tx.clone(); + let options: Options = options.clone(); + + async move { + let (response, upgrade) = WebSocket::upgrade_with_options(&mut request, options)?; + + tokio::spawn(async move { + match upgrade.await { + Ok(socket) => { + let _ = upgraded_tx.send(socket).await; + } + Err(e) => tracing::warn!("WebSocket upgrade failed: {e}"), + } + }); + + Ok::<_, yawc::WebSocketError>(response) + } + }); + // Bound clients that open TCP but never complete the WebSocket handshake. - let ws_stream = tokio::time::timeout( - self.websocket_handshake_timeout, - native::accept_async_with_config(raw_stream, Some(self.websocket_config())), - ) + let socket = tokio::time::timeout(self.websocket_handshake_timeout, async { + http1::Builder::new() + .serve_connection(TokioIo::new(raw_stream), service) + .with_upgrades() + .await + .map_err(Error::transport)?; + + upgraded_rx.recv().await.ok_or_else(|| { + Error::with_static_message(ErrorKind::Transport, "WebSocket upgrade failed") + }) + }) .await .map_err(|_| { Error::with_static_message(ErrorKind::Transport, "WebSocket handshake timed out") - })? - .map_err(Error::transport)?; + })??; // The pre-handshake socket is no longer consuming admission resources. drop(permit); @@ -329,7 +365,8 @@ impl InnerLocalRelay { // An established connection only consumes a permit when explicitly configured. let permit = self.connections_limit.clone().try_acquire_owned()?; - self.handle_websocket(ws_stream, addr, permit).await?; + self.handle_websocket(Reporting::new(socket), addr, permit) + .await?; Ok(()) } @@ -337,7 +374,7 @@ impl InnerLocalRelay { /// Handle websocket connection async fn handle_websocket( &self, - ws_stream: WebSocketStream, + ws_stream: Reporting, addr: SocketAddr, _permit: OwnedSemaphorePermit, ) -> Result<(), Error> @@ -374,11 +411,12 @@ impl InnerLocalRelay { return Err(Error::limit_exceeded("too many client messages")); } - match msg { - Message::Text(json) => { - tracing::trace!("Received {json}"); - let message_size = json.len(); - match ClientMessage::from_json(json.as_bytes()) { + match msg.opcode() { + OpCode::Text => { + let payload = msg.payload(); + tracing::trace!("Received {}", String::from_utf8_lossy(payload)); + let message_size = payload.len(); + match ClientMessage::from_json(payload.as_ref()) { Ok(msg) => { self.handle_client_msg( &mut session, @@ -401,17 +439,17 @@ impl InnerLocalRelay { } } } - Message::Binary(..) => { + OpCode::Binary => { let msg = RelayMessage::Notice(Cow::Borrowed("binary messages are not processed by this relay")); if let Err(e) = send_msg(&mut tx, msg).await { tracing::error!("Can't send msg to client: {e}"); } } - Message::Ping(..) => {} - Message::Pong(..) => {} - Message::Close(..) => {} - Message::Frame(..) => {} + OpCode::Continuation + | OpCode::Ping + | OpCode::Pong + | OpCode::Close => {} } } Some(Err(e)) => tracing::error!("Can't handle websocket msg: {e}"), @@ -447,10 +485,13 @@ impl InnerLocalRelay { Ok(()) } - fn websocket_config(&self) -> WebSocketConfig { - WebSocketConfig::default() - .max_message_size(Some(self.max_websocket_message_size)) - .max_frame_size(Some(self.max_websocket_message_size)) + fn websocket_options(&self) -> Options { + Options::default() + .with_limits( + self.max_websocket_message_size, + self.max_websocket_message_size, + ) + .with_utf8() } async fn handle_client_msg( @@ -1480,7 +1521,7 @@ async fn send_msg(tx: &mut WsTx, msg: RelayMessage<'_>) -> Result<(), Erro where S: AsyncRead + AsyncWrite + Unpin, { - tx.send(Message::Text(msg.as_json().into())) + tx.send(Frame::text(msg.as_json())) .await .map_err(|e| Error::new(ErrorKind::Other, e))?; Ok(()) @@ -1720,7 +1761,7 @@ mod tests { #[test] fn local_relay_defaults_bound_connection_resources() { let relay = InnerLocalRelay::new(LocalRelayBuilder::default()); - let config = relay.websocket_config(); + let options = relay.websocket_options(); assert_eq!(relay.pending_handshakes_limit.available_permits(), 128); assert_eq!( @@ -1734,8 +1775,8 @@ mod tests { assert_eq!(relay.max_filter_limit, 500); assert_eq!(relay.max_subscription_bytes, 1024 * 1024); assert_eq!(relay.max_negentropy_items, 50_000); - assert_eq!(config.max_message_size, Some(5 * 1024 * 1024)); - assert_eq!(config.max_frame_size, Some(5 * 1024 * 1024)); + assert_eq!(options.max_payload_read, Some(5 * 1024 * 1024)); + assert_eq!(options.max_read_buffer, Some(5 * 1024 * 1024)); assert_eq!(relay.websocket_handshake_timeout.as_secs(), 10); } @@ -1747,12 +1788,12 @@ mod tests { .max_websocket_message_size(1024) .websocket_handshake_timeout(Duration::from_secs(2)), ); - let config = relay.websocket_config(); + let options = relay.websocket_options(); assert_eq!(relay.pending_handshakes_limit.available_permits(), 128); assert_eq!(relay.connections_limit.available_permits(), 4); - assert_eq!(config.max_message_size, Some(1024)); - assert_eq!(config.max_frame_size, Some(1024)); + assert_eq!(options.max_payload_read, Some(1024)); + assert_eq!(options.max_read_buffer, Some(1024)); assert_eq!(relay.websocket_handshake_timeout.as_secs(), 2); } @@ -1781,9 +1822,11 @@ mod tests { relay.database.save_event(&event).await.unwrap(); let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = WebSocketStream::from_raw_socket(server_stream, Role::Server, None).await; - let mut client = WebSocketStream::from_raw_socket(client_stream, Role::Client, None).await; - let (mut server_tx, _) = server.split(); + let server = + WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); + let mut client = + WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); + let (mut server_tx, _) = Reporting::new(server).split(); let mut session = session(None); let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); @@ -1799,12 +1842,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::Ok { status: false, message, @@ -1821,9 +1862,11 @@ mod tests { .query_policy(ReplaceWithGiftWrap), ); let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = WebSocketStream::from_raw_socket(server_stream, Role::Server, None).await; - let mut client = WebSocketStream::from_raw_socket(client_stream, Role::Client, None).await; - let (mut server_tx, _) = server.split(); + let server = + WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); + let mut client = + WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); + let (mut server_tx, _) = Reporting::new(server).split(); let mut session = session(None); let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); @@ -1841,24 +1884,18 @@ mod tests { .await .unwrap(); - let auth = client.next().await.unwrap().unwrap(); - let closed = client.next().await.unwrap().unwrap(); + let auth = client.next().await.unwrap(); + let closed = client.next().await.unwrap(); + assert_eq!(auth.opcode(), OpCode::Text); assert!(matches!( - auth, - Message::Text(json) - if matches!( - RelayMessage::from_json(json.as_bytes()).unwrap(), - RelayMessage::Auth { .. } - ) + RelayMessage::from_json(auth.payload()).unwrap(), + RelayMessage::Auth { .. } )); + assert_eq!(closed.opcode(), OpCode::Text); assert!(matches!( - closed, - Message::Text(json) - if matches!( - RelayMessage::from_json(json.as_bytes()).unwrap(), - RelayMessage::Closed { message, .. } - if message.starts_with("auth-required:") - ) + RelayMessage::from_json(closed.payload()).unwrap(), + RelayMessage::Closed { message, .. } + if message.starts_with("auth-required:") )); } @@ -1866,9 +1903,11 @@ mod tests { async fn oversized_active_subscription_is_rejected() { let relay = InnerLocalRelay::new(LocalRelayBuilder::default().max_subscription_bytes(10)); let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = WebSocketStream::from_raw_socket(server_stream, Role::Server, None).await; - let mut client = WebSocketStream::from_raw_socket(client_stream, Role::Client, None).await; - let (mut server_tx, _) = server.split(); + let server = + WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); + let mut client = + WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); + let (mut server_tx, _) = Reporting::new(server).split(); let mut session = session(None); let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); @@ -1886,12 +1925,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::Closed { subscription_id, message, @@ -1906,9 +1943,11 @@ mod tests { async fn excessive_req_filters_are_rejected() { let relay = InnerLocalRelay::new(LocalRelayBuilder::default().max_filters_per_req(1)); let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = WebSocketStream::from_raw_socket(server_stream, Role::Server, None).await; - let mut client = WebSocketStream::from_raw_socket(client_stream, Role::Client, None).await; - let (mut server_tx, _) = server.split(); + let server = + WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); + let mut client = + WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); + let (mut server_tx, _) = Reporting::new(server).split(); let mut session = session(None); let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); @@ -1926,12 +1965,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::Closed { subscription_id, message, @@ -1944,9 +1981,11 @@ mod tests { async fn zero_query_rate_rejects_query_starts() { let relay = InnerLocalRelay::new(LocalRelayBuilder::default().queries_per_minute(0)); let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = WebSocketStream::from_raw_socket(server_stream, Role::Server, None).await; - let mut client = WebSocketStream::from_raw_socket(client_stream, Role::Client, None).await; - let (mut server_tx, _) = server.split(); + let server = + WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); + let mut client = + WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); + let (mut server_tx, _) = Reporting::new(server).split(); let mut session = session(None); session.query_tokens = Tokens::new(0); session.negentropy_tokens = Tokens::new(0); @@ -1966,12 +2005,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::Closed { subscription_id, message, @@ -1993,12 +2030,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -2020,12 +2055,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::Closed { subscription_id, message, @@ -2048,12 +2081,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -2066,9 +2097,11 @@ mod tests { async fn negentropy_continuations_do_not_consume_query_start_allowance() { let relay = InnerLocalRelay::new(LocalRelayBuilder::default().queries_per_minute(1)); let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = WebSocketStream::from_raw_socket(server_stream, Role::Server, None).await; - let mut client = WebSocketStream::from_raw_socket(client_stream, Role::Client, None).await; - let (mut server_tx, _) = server.split(); + let server = + WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); + let mut client = + WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); + let (mut server_tx, _) = Reporting::new(server).split(); let mut session = session(None); let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); @@ -2086,12 +2119,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -2115,12 +2146,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -2143,12 +2172,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::EndOfStoredEvents(subscription_id) if subscription_id.as_str() == "allowed" )); @@ -2168,12 +2195,10 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); - let Message::Text(response) = response else { - panic!("unexpected WebSocket message"); - }; + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(response.as_bytes()).unwrap(), + RelayMessage::from_json(response.payload()).unwrap(), RelayMessage::Closed { subscription_id, message, @@ -2189,9 +2214,11 @@ mod tests { .finalize(&Keys::generate()) .unwrap(); let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = WebSocketStream::from_raw_socket(server_stream, Role::Server, None).await; - let mut client = WebSocketStream::from_raw_socket(client_stream, Role::Client, None).await; - let (mut server_tx, _) = server.split(); + let server = + WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); + let mut client = + WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); + let (mut server_tx, _) = Reporting::new(server).split(); let mut session = session(None); session.auth_tokens = Tokens::new(0); let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); @@ -2207,15 +2234,12 @@ mod tests { .await .unwrap(); - let response = client.next().await.unwrap().unwrap(); + let response = client.next().await.unwrap(); + assert_eq!(response.opcode(), OpCode::Text); assert!(matches!( - response, - Message::Text(json) - if matches!( - RelayMessage::from_json(json.as_bytes()).unwrap(), - RelayMessage::Ok { status: false, message, .. } - if message == "rate-limited: too many authentication attempts" - ) + RelayMessage::from_json(response.payload()).unwrap(), + RelayMessage::Ok { status: false, message, .. } + if message == "rate-limited: too many authentication attempts" )); } diff --git a/nostr-sdk/src/local_relay/local/mod.rs b/nostr-sdk/src/local_relay/local/mod.rs index 525d4acea..96cb407f8 100644 --- a/nostr-sdk/src/local_relay/local/mod.rs +++ b/nostr-sdk/src/local_relay/local/mod.rs @@ -158,14 +158,16 @@ mod tests { use std::pin::Pin; use std::time::Duration; - use async_wsocket::{ConnectionMode, Message, Url, WebSocket}; use futures::{SinkExt, StreamExt}; use negentropy::{Negentropy, NegentropyStorageVector}; use nostr::event::{EventBuilder, FinalizeEvent, Kind}; use nostr::filter::Filter; use nostr::key::Keys; use nostr::message::{MachineReadablePrefix, RelayMessage}; + use nostr::types::Url; use tokio::time; + use yawc::WebSocket; + use yawc::frame::{Frame, OpCode}; use super::*; use crate::local_relay::{QueryPolicy, QueryPolicyResult}; @@ -216,18 +218,16 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); socket - .send(Message::Text( - r#"["REQ","short-author",{"authors":["deadbeef"]}]"#.to_owned(), + .send(Frame::text( + r#"["REQ","short-author",{"authors":["deadbeef"]}]"#, )) .await .unwrap(); socket - .send(Message::Text(r#"["REQ","valid",{}]"#.to_owned())) + .send(Frame::text(r#"["REQ","valid",{}]"#)) .await .unwrap(); @@ -238,11 +238,10 @@ mod tests { let message = socket .next() .await - .expect("WebSocket connection terminated") - .unwrap(); + .expect("WebSocket connection terminated"); - if let Message::Text(json) = message { - match RelayMessage::from_json(json.as_bytes()).unwrap() { + if message.opcode() == OpCode::Text { + match RelayMessage::from_json(message.payload()).unwrap() { RelayMessage::Notice(..) => received_notice = true, RelayMessage::EndOfStoredEvents(subscription_id) if subscription_id.as_str() == "valid" => @@ -265,28 +264,22 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); socket - .send(Message::Text( - r#"["NEG-OPEN","neg-odd",{},"abc"]"#.to_owned(), - )) + .send(Frame::text(r#"["NEG-OPEN","neg-odd",{},"abc"]"#)) .await .unwrap(); socket - .send(Message::Text( - r#"["NEG-OPEN","neg-nonhex",{},"zz"]"#.to_owned(), - )) + .send(Frame::text(r#"["NEG-OPEN","neg-nonhex",{},"zz"]"#)) .await .unwrap(); socket - .send(Message::Text(r#"["NEG-MSG","neg-msg","abc"]"#.to_owned())) + .send(Frame::text(r#"["NEG-MSG","neg-msg","abc"]"#)) .await .unwrap(); socket - .send(Message::Text(r#"["REQ","valid",{}]"#.to_owned())) + .send(Frame::text(r#"["REQ","valid",{}]"#)) .await .unwrap(); @@ -297,11 +290,10 @@ mod tests { let message = socket .next() .await - .expect("WebSocket connection terminated") - .unwrap(); + .expect("WebSocket connection terminated"); - if let Message::Text(json) = message { - match RelayMessage::from_json(json.as_bytes()).unwrap() { + if message.opcode() == OpCode::Text { + match RelayMessage::from_json(message.payload()).unwrap() { RelayMessage::NegErr { message, .. } => { assert_eq!(message, "error: invalid negentropy message"); neg_errors += 1; @@ -327,9 +319,7 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); // Open a valid negentropy subscription let mut storage = NegentropyStorageVector::new(); @@ -337,33 +327,29 @@ mod tests { let mut negentropy = Negentropy::owned(storage, 60_000).unwrap(); let initial_message = faster_hex::hex_string(&negentropy.initiate().unwrap()); socket - .send(Message::Text(format!( + .send(Frame::text(format!( r#"["NEG-OPEN","neg",{{}},"{initial_message}"]"# ))) .await .unwrap(); - let reply = socket.next().await.unwrap().unwrap(); - let Message::Text(reply) = reply else { - panic!("unexpected websocket message"); - }; + let reply = socket.next().await.unwrap(); + assert_eq!(reply.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(reply.as_bytes()).unwrap(), + RelayMessage::from_json(reply.payload()).unwrap(), RelayMessage::NegMsg { .. } )); // A malformed payload must terminate only this negentropy subscription socket - .send(Message::Text(r#"["NEG-MSG","neg","zz"]"#.to_owned())) + .send(Frame::text(r#"["NEG-MSG","neg","zz"]"#)) .await .unwrap(); - let neg_err = socket.next().await.unwrap().unwrap(); - let Message::Text(neg_err) = neg_err else { - panic!("unexpected websocket message"); - }; + let neg_err = socket.next().await.unwrap(); + assert_eq!(neg_err.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + RelayMessage::from_json(neg_err.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -373,16 +359,14 @@ mod tests { // The subscription is gone, but the connection keeps serving requests socket - .send(Message::Text(r#"["NEG-MSG","neg","6100"]"#.to_owned())) + .send(Frame::text(r#"["NEG-MSG","neg","6100"]"#)) .await .unwrap(); - let neg_err = socket.next().await.unwrap().unwrap(); - let Message::Text(neg_err) = neg_err else { - panic!("unexpected websocket message"); - }; + let neg_err = socket.next().await.unwrap(); + assert_eq!(neg_err.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + RelayMessage::from_json(neg_err.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -391,16 +375,14 @@ mod tests { )); socket - .send(Message::Text(r#"["REQ","valid",{}]"#.to_owned())) + .send(Frame::text(r#"["REQ","valid",{}]"#)) .await .unwrap(); - let eose = socket.next().await.unwrap().unwrap(); - let Message::Text(eose) = eose else { - panic!("unexpected websocket message"); - }; + let eose = socket.next().await.unwrap(); + assert_eq!(eose.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(eose.as_bytes()).unwrap(), + RelayMessage::from_json(eose.payload()).unwrap(), RelayMessage::EndOfStoredEvents(subscription_id) if subscription_id.as_str() == "valid" )); @@ -414,29 +396,23 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); socket - .send(Message::Text(r#"["COUNT","count",{}]"#.to_owned())) + .send(Frame::text(r#"["COUNT","count",{}]"#)) .await .unwrap(); - let auth = socket.next().await.unwrap().unwrap(); - let Message::Text(auth) = auth else { - panic!("unexpected websocket message"); - }; + let auth = socket.next().await.unwrap(); + assert_eq!(auth.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(auth.as_bytes()).unwrap(), + RelayMessage::from_json(auth.payload()).unwrap(), RelayMessage::Auth { .. } )); - let closed = socket.next().await.unwrap().unwrap(); - let Message::Text(closed) = closed else { - panic!("unexpected websocket message"); - }; + let closed = socket.next().await.unwrap(); + assert_eq!(closed.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(closed.as_bytes()).unwrap(), + RelayMessage::from_json(closed.payload()).unwrap(), RelayMessage::Closed { subscription_id, message, @@ -453,29 +429,23 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); socket - .send(Message::Text(r#"["NEG-OPEN","neg",{},""]"#.to_owned())) + .send(Frame::text(r#"["NEG-OPEN","neg",{},""]"#)) .await .unwrap(); - let auth = socket.next().await.unwrap().unwrap(); - let Message::Text(auth) = auth else { - panic!("unexpected websocket message"); - }; + let auth = socket.next().await.unwrap(); + assert_eq!(auth.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(auth.as_bytes()).unwrap(), + RelayMessage::from_json(auth.payload()).unwrap(), RelayMessage::Auth { .. } )); - let neg_err = socket.next().await.unwrap().unwrap(); - let Message::Text(neg_err) = neg_err else { - panic!("unexpected websocket message"); - }; + let neg_err = socket.next().await.unwrap(); + assert_eq!(neg_err.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + RelayMessage::from_json(neg_err.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -490,20 +460,16 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); socket - .send(Message::Text(r#"["COUNT","count",{}]"#.to_owned())) + .send(Frame::text(r#"["COUNT","count",{}]"#)) .await .unwrap(); - let closed = socket.next().await.unwrap().unwrap(); - let Message::Text(closed) = closed else { - panic!("unexpected websocket message"); - }; + let closed = socket.next().await.unwrap(); + assert_eq!(closed.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(closed.as_bytes()).unwrap(), + RelayMessage::from_json(closed.payload()).unwrap(), RelayMessage::Closed { subscription_id, message, @@ -512,16 +478,14 @@ mod tests { )); socket - .send(Message::Text(r#"["NEG-OPEN","neg",{},""]"#.to_owned())) + .send(Frame::text(r#"["NEG-OPEN","neg",{},""]"#)) .await .unwrap(); - let neg_err = socket.next().await.unwrap().unwrap(); - let Message::Text(neg_err) = neg_err else { - panic!("unexpected websocket message"); - }; + let neg_err = socket.next().await.unwrap(); + assert_eq!(neg_err.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + RelayMessage::from_json(neg_err.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -538,20 +502,16 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); socket - .send(Message::Text(r#"["NEG-OPEN","neg",{},""]"#.to_owned())) + .send(Frame::text(r#"["NEG-OPEN","neg",{},""]"#)) .await .unwrap(); - let neg_err = socket.next().await.unwrap().unwrap(); - let Message::Text(neg_err) = neg_err else { - panic!("unexpected websocket message"); - }; + let neg_err = socket.next().await.unwrap(); + assert_eq!(neg_err.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + RelayMessage::from_json(neg_err.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -566,20 +526,16 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); socket - .send(Message::Text(r#"["NEG-OPEN","éé",{},""]"#.to_owned())) + .send(Frame::text(r#"["NEG-OPEN","éé",{},""]"#)) .await .unwrap(); - let neg_err = socket.next().await.unwrap().unwrap(); - let Message::Text(neg_err) = neg_err else { - panic!("unexpected websocket message"); - }; + let neg_err = socket.next().await.unwrap(); + assert_eq!(neg_err.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + RelayMessage::from_json(neg_err.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -605,20 +561,16 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); socket - .send(Message::Text(r#"["NEG-OPEN","neg",{},""]"#.to_owned())) + .send(Frame::text(r#"["NEG-OPEN","neg",{},""]"#)) .await .unwrap(); - let neg_err = socket.next().await.unwrap().unwrap(); - let Message::Text(neg_err) = neg_err else { - panic!("unexpected websocket message"); - }; + let neg_err = socket.next().await.unwrap(); + assert_eq!(neg_err.opcode(), OpCode::Text); assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + RelayMessage::from_json(neg_err.payload()).unwrap(), RelayMessage::NegErr { subscription_id, message, @@ -633,31 +585,26 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); time::timeout(Duration::from_secs(5), async { for _ in 0..301 { socket - .send(Message::Text(r#"["CLOSE","harmless"]"#.to_owned())) + .send(Frame::text(r#"["CLOSE","harmless"]"#)) .await .unwrap(); } socket - .send(Message::Text(r#"["REQ","valid",{}]"#.to_owned())) + .send(Frame::text(r#"["REQ","valid",{}]"#)) .await .unwrap(); - let eose = socket.next().await.unwrap().unwrap(); + let eose = socket.next().await.unwrap(); + assert_eq!(eose.opcode(), OpCode::Text); assert!(matches!( - eose, - Message::Text(json) - if matches!( - RelayMessage::from_json(json.as_bytes()).unwrap(), - RelayMessage::EndOfStoredEvents(subscription_id) - if subscription_id.as_str() == "valid" - ) + RelayMessage::from_json(eose.payload()).unwrap(), + RelayMessage::EndOfStoredEvents(subscription_id) + if subscription_id.as_str() == "valid" )); }) .await @@ -670,23 +617,18 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); time::timeout(Duration::from_secs(5), async { for _ in 0..6_001 { socket - .send(Message::Text(r#"["CLOSE","harmless"]"#.to_owned())) + .send(Frame::text(r#"["CLOSE","harmless"]"#)) .await .unwrap(); } let closed = socket.next().await; - assert!(matches!( - closed, - None | Some(Ok(Message::Close(..))) | Some(Err(..)) - )); + assert!(closed.is_none_or(|frame| frame.opcode() == OpCode::Close)); }) .await .expect("connection did not close after 6,001 rapid frames"); @@ -698,30 +640,22 @@ mod tests { relay.run().await.unwrap(); let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(&url, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = WebSocket::connect(url).await.unwrap(); time::timeout(Duration::from_secs(5), async { for byte in 1..=3 { - socket.send(Message::Binary(vec![byte])).await.unwrap(); - let notice = socket.next().await.unwrap().unwrap(); + socket.send(Frame::binary(vec![byte])).await.unwrap(); + let notice = socket.next().await.unwrap(); + assert_eq!(notice.opcode(), OpCode::Text); assert!(matches!( - notice, - Message::Text(json) - if matches!( - RelayMessage::from_json(json.as_bytes()).unwrap(), - RelayMessage::Notice(..) - ) + RelayMessage::from_json(notice.payload()).unwrap(), + RelayMessage::Notice(..) )); } - socket.send(Message::Binary(vec![4])).await.unwrap(); + socket.send(Frame::binary(vec![4])).await.unwrap(); let closed = socket.next().await; - assert!(matches!( - closed, - None | Some(Ok(Message::Close(..))) | Some(Err(..)) - )); + assert!(closed.is_none_or(|frame| frame.opcode() == OpCode::Close)); }) .await .expect("connection did not enforce the configured three-frame limit"); diff --git a/nostr-sdk/src/transport/websocket.rs b/nostr-sdk/src/transport/websocket.rs index 63a5ffbbe..41f5a0a98 100644 --- a/nostr-sdk/src/transport/websocket.rs +++ b/nostr-sdk/src/transport/websocket.rs @@ -162,7 +162,7 @@ where /// broken connection indistinguishable from one the relay closed cleanly. Polling the frame /// directly keeps the two apart, at the cost of tracking the terminal state here. #[cfg(not(target_arch = "wasm32"))] -struct Reporting { +pub(crate) struct Reporting { socket: WebSocket, /// Set once a read has failed, so the socket is not polled again afterwards. failed: bool, @@ -170,7 +170,7 @@ struct Reporting { #[cfg(not(target_arch = "wasm32"))] impl Reporting { - fn new(socket: WebSocket) -> Self { + pub(crate) fn new(socket: WebSocket) -> Self { Self { socket, failed: false, From 6873dda8230e446122f56a59b509ba3fbaa97dcb Mon Sep 17 00:00:00 2001 From: Dario Date: Wed, 9 Sep 2026 11:22:08 +0200 Subject: [PATCH 3/7] sdk: update changelogs for the yawc migration Record the transport's move to yawc, the frame type that comes with it, the TLS features that no longer have an equivalent, and the local relay now serving its handshake over hyper. Note the removed features in the nwc and nostr-connect changelogs too, since both forwarded them. --- nostr-sdk/CHANGELOG.md | 19 +++++++++++++++++++ nwc/CHANGELOG.md | 7 +++++++ signer/nostr-connect/CHANGELOG.md | 7 +++++++ 3 files changed, 33 insertions(+) diff --git a/nostr-sdk/CHANGELOG.md b/nostr-sdk/CHANGELOG.md index 822d77b4f..0b0800977 100644 --- a/nostr-sdk/CHANGELOG.md +++ b/nostr-sdk/CHANGELOG.md @@ -29,6 +29,25 @@ ## Unreleased +### Breaking changes + +- Replace `async-wsocket` with `yawc` as the WebSocket implementation. The transport + sink and stream now carry `yawc::frame::Frame` instead of `async_wsocket::Message`, + so custom `WebSocketTransport` implementations build frames with `Frame::text` and + match incoming ones on `Frame::opcode` +- Remove the `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` + features, which have no equivalent in `yawc`. `ring` and `aws_lc_rs` now select + yawc's rustls backends, and `rustls-tls-webpki-roots` is kept as a no-op because + yawc always trusts the webpki roots. The same features are removed from `nwc` and + `nostr-connect` + +### Changed + +- Route proxied relay connections through `socks5h`, leaving name resolution to the + proxy so `.onion` addresses stay reachable +- Serve the local relay's listener over hyper, which is how yawc reaches its server + side. `LocalRelay::take_connection` is unchanged + ### Deprecated - Deprecate `LocalRelayBuilder::max_query_results` and `LocalRelayBuilder::default_filter_limit` (https://github.com/nostrdevkit/nostr/pull/1461) diff --git a/nwc/CHANGELOG.md b/nwc/CHANGELOG.md index 55574f782..a0bb643e8 100644 --- a/nwc/CHANGELOG.md +++ b/nwc/CHANGELOG.md @@ -27,6 +27,13 @@ --> +## Unreleased + +### Breaking changes + +- Remove the `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` features, + following their removal from `nostr-sdk` + ## v0.45.0 - 2026/08/05 ### Breaking changes diff --git a/signer/nostr-connect/CHANGELOG.md b/signer/nostr-connect/CHANGELOG.md index 16ec8527c..216aa65af 100644 --- a/signer/nostr-connect/CHANGELOG.md +++ b/signer/nostr-connect/CHANGELOG.md @@ -27,6 +27,13 @@ --> +## Unreleased + +### Breaking changes + +- Remove the `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` features, + following their removal from `nostr-sdk` + ## v0.45.1 - 2026/08/07 ### Fixed From 1fa065b01c508c04f6a7120c59870d2e01e7d098 Mon Sep 17 00:00:00 2001 From: Dario Date: Wed, 9 Sep 2026 14:54:15 +0200 Subject: [PATCH 4/7] sdk: fix a stale doc comment in the hyper example The helper's docs still pointed at tungstenite's `from_raw_socket`, which no longer exists here. Say what the helper is actually for instead. --- nostr-sdk/examples/local-relay-hyper.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/nostr-sdk/examples/local-relay-hyper.rs b/nostr-sdk/examples/local-relay-hyper.rs index 3463df46c..bd3d1144b 100644 --- a/nostr-sdk/examples/local-relay-hyper.rs +++ b/nostr-sdk/examples/local-relay-hyper.rs @@ -23,11 +23,11 @@ struct HttpServer { remote: SocketAddr, } -/// Copied from https://github.com/snapview/tungstenite-rs/blob/c16778797b2eeb118aa064aa5b483f90c3989627/src/handshake/mod.rs#L112C1-L125C1 /// Derive the `Sec-WebSocket-Accept` response header from a `Sec-WebSocket-Key` request header. /// -/// This function can be used to perform a handshake before passing a raw TCP stream to -/// [`WebSocket::from_raw_socket`][crate::protocol::WebSocket::from_raw_socket]. +/// This example completes the handshake itself so it can hand the upgraded stream to +/// [`LocalRelay::take_connection`], which is the entry point for servers that already +/// speak HTTP. pub fn derive_accept_key(request_key: &[u8]) -> String { // ... field is constructed by concatenating /key/ ... // ... with the string "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" (RFC 6455) From 3fcd783fee673da6b34ea51818593c16319753a0 Mon Sep 17 00:00:00 2001 From: Dario Date: Thu, 24 Sep 2026 09:26:08 +0200 Subject: [PATCH 5/7] sdk: propagate WebSocket test server errors --- nostr-sdk/src/transport/websocket.rs | 67 +++++++++++++++------------- 1 file changed, 36 insertions(+), 31 deletions(-) diff --git a/nostr-sdk/src/transport/websocket.rs b/nostr-sdk/src/transport/websocket.rs index 41f5a0a98..3dac808dc 100644 --- a/nostr-sdk/src/transport/websocket.rs +++ b/nostr-sdk/src/transport/websocket.rs @@ -263,13 +263,16 @@ where #[cfg(all(test, not(target_arch = "wasm32")))] mod tests { + use std::error::Error as StdError; + use std::io; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; use super::*; /// Read the HTTP request a client wrote, up to the blank line that ends its head. - async fn read_request(stream: &mut S) -> String + async fn read_request(stream: &mut S) -> io::Result where S: tokio::io::AsyncRead + Unpin, { @@ -277,58 +280,61 @@ mod tests { let mut len = 0; while len < request.len() && !request[..len].ends_with(b"\r\n\r\n") { - let read = stream.read(&mut request[len..]).await.unwrap(); + let read = stream.read(&mut request[len..]).await?; if read == 0 { break; } len += read; } - String::from_utf8(request[..len].to_vec()).unwrap() + String::from_utf8(request[..len].to_vec()) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) } #[tokio::test] - async fn default_transport_sends_user_agent() { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let address = listener.local_addr().unwrap(); + async fn default_transport_sends_user_agent() -> Result<(), Box> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; let server = tokio::spawn(async move { - let (mut stream, _) = listener.accept().await.unwrap(); + let (mut stream, _) = listener.accept().await?; read_request(&mut stream).await }); - let url = Url::parse(&format!("ws://{address}")).unwrap(); + let url = Url::parse(&format!("ws://{address}"))?; assert!(DefaultWebsocketTransport.connect(&url, None).await.is_err()); - let request = server.await.unwrap(); + let request = server.await??; let user_agent = request.lines().find_map(|line| { let (name, value) = line.split_once(':')?; name.eq_ignore_ascii_case("user-agent") .then(|| value.trim()) }); assert_eq!(user_agent, Some(USER_AGENT)); + + Ok(()) } /// Serve one SOCKS5 CONNECT, then read whatever the client tunnels through it. /// /// Nothing answers the WebSocket handshake, so the client's connect fails. What the test /// cares about is the address the proxy was asked to reach and the bytes that followed. - async fn socks5_connect(listener: TcpListener) -> (String, String) { - let (mut stream, _) = listener.accept().await.unwrap(); + async fn socks5_connect(listener: TcpListener) -> io::Result<(String, String)> { + let (mut stream, _) = listener.accept().await?; // Greeting: version, then the authentication methods the client offers. let mut greeting = [0u8; 2]; - stream.read_exact(&mut greeting).await.unwrap(); + stream.read_exact(&mut greeting).await?; assert_eq!(greeting[0], 0x05); let mut methods = vec![0u8; greeting[1] as usize]; - stream.read_exact(&mut methods).await.unwrap(); + stream.read_exact(&mut methods).await?; // "No authentication required" - stream.write_all(&[0x05, 0x00]).await.unwrap(); + stream.write_all(&[0x05, 0x00]).await?; // Request: version, command, reserved, address type. let mut request = [0u8; 4]; - stream.read_exact(&mut request).await.unwrap(); + stream.read_exact(&mut request).await?; assert_eq!(request[0], 0x05); assert_eq!(request[1], 0x01, "expected a CONNECT command"); assert_eq!( @@ -337,37 +343,34 @@ mod tests { ); let mut length = [0u8; 1]; - stream.read_exact(&mut length).await.unwrap(); + stream.read_exact(&mut length).await?; let mut domain = vec![0u8; length[0] as usize]; - stream.read_exact(&mut domain).await.unwrap(); + stream.read_exact(&mut domain).await?; let mut port = [0u8; 2]; - stream.read_exact(&mut port).await.unwrap(); + stream.read_exact(&mut port).await?; - let target: String = format!( - "{}:{}", - String::from_utf8(domain).unwrap(), - u16::from_be_bytes(port) - ); + let domain = + String::from_utf8(domain).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + let target: String = format!("{domain}:{}", u16::from_be_bytes(port)); // Success, with a bound address of 0.0.0.0:0. stream .write_all(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) - .await - .unwrap(); + .await?; - (target, read_request(&mut stream).await) + Ok((target, read_request(&mut stream).await?)) } #[tokio::test] - async fn default_transport_dials_through_socks5_proxy() { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let proxy = listener.local_addr().unwrap(); + async fn default_transport_dials_through_socks5_proxy() -> Result<(), Box> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let proxy = listener.local_addr()?; let server = tokio::spawn(socks5_connect(listener)); // A name that never resolves locally, so reaching it proves the proxy resolved it. // This is what makes `.onion` addresses work. - let url = Url::parse("ws://relay.invalid:8080").unwrap(); + let url = Url::parse("ws://relay.invalid:8080")?; assert!( DefaultWebsocketTransport .connect(&url, Some(proxy)) @@ -375,8 +378,10 @@ mod tests { .is_err() ); - let (target, request) = server.await.unwrap(); + let (target, request) = server.await??; assert_eq!(target, "relay.invalid:8080"); assert!(request.starts_with("GET / HTTP/1.1"), "{request}"); + + Ok(()) } } From 3047b01d1b4125d87b0a964d465c0c602a6138c8 Mon Sep 17 00:00:00 2001 From: Dario Date: Thu, 24 Sep 2026 10:11:22 +0200 Subject: [PATCH 6/7] refactor: simplify WebSocket upgrade handling Pass the pending upgrade through the existing channel and await it after Hyper writes the response. This avoids spawning a task per connection while preserving the original upgrade error. --- nostr-sdk/src/local_relay/local/inner.rs | 47 ++++++++++-------------- nostr-sdk/src/transport/websocket.rs | 12 +++--- 2 files changed, 25 insertions(+), 34 deletions(-) diff --git a/nostr-sdk/src/local_relay/local/inner.rs b/nostr-sdk/src/local_relay/local/inner.rs index 24affca29..18f28e6af 100644 --- a/nostr-sdk/src/local_relay/local/inner.rs +++ b/nostr-sdk/src/local_relay/local/inner.rs @@ -298,9 +298,7 @@ impl InnerLocalRelay { .map_err(Error::transport)?; self.handle_websocket(Reporting::new(socket), addr, permit) - .await?; - - Ok(()) + .await } /// Pass bare [TcpStream] for handling @@ -318,29 +316,20 @@ impl InnerLocalRelay { } // Take the connection permit before doing the handshake. - let connection_permit: OwnedSemaphorePermit = - self.connections_limit.clone().try_acquire_owned()?; + let connection_permit = self.connections_limit.clone().try_acquire_owned()?; - // The upgraded socket only exists once the response has been written, so the service - // hands it back over a channel instead of returning it. - let (upgraded_tx, mut upgraded_rx) = mpsc::channel(1); - let options: Options = self.websocket_options(); + // The upgrade future only resolves once the response has been written, so the service + // hands it back over a channel instead of awaiting it. + let (upgrade_tx, mut upgrade_rx) = mpsc::channel(1); + let options = self.websocket_options(); let service = service_fn(move |mut request: hyper::Request| { - let upgraded_tx = upgraded_tx.clone(); - let options: Options = options.clone(); + let upgrade_tx = upgrade_tx.clone(); + let options = options.clone(); async move { let (response, upgrade) = WebSocket::upgrade_with_options(&mut request, options)?; - - tokio::spawn(async move { - match upgrade.await { - Ok(socket) => { - let _ = upgraded_tx.send(socket).await; - } - Err(e) => tracing::warn!("WebSocket upgrade failed: {e}"), - } - }); + let _ = upgrade_tx.send(upgrade).await; Ok::<_, yawc::WebSocketError>(response) } @@ -354,9 +343,14 @@ impl InnerLocalRelay { .await .map_err(Error::transport)?; - upgraded_rx.recv().await.ok_or_else(|| { - Error::with_static_message(ErrorKind::Transport, "WebSocket upgrade failed") - }) + upgrade_rx + .recv() + .await + .ok_or_else(|| { + Error::with_static_message(ErrorKind::Transport, "WebSocket upgrade failed") + })? + .await + .map_err(Error::transport) }) .await .map_err(|_| { @@ -367,9 +361,7 @@ impl InnerLocalRelay { drop(handshake_permit); self.handle_websocket(Reporting::new(socket), addr, connection_permit) - .await?; - - Ok(()) + .await } /// Handle websocket connection @@ -1557,8 +1549,7 @@ where { tx.send(Frame::text(msg.as_json())) .await - .map_err(|e| Error::new(ErrorKind::Other, e))?; - Ok(()) + .map_err(Error::other) } async fn send_query_rate_limit_error( diff --git a/nostr-sdk/src/transport/websocket.rs b/nostr-sdk/src/transport/websocket.rs index 3dac808dc..b1e5057ed 100644 --- a/nostr-sdk/src/transport/websocket.rs +++ b/nostr-sdk/src/transport/websocket.rs @@ -98,7 +98,7 @@ impl WebSocketTransport for DefaultWebsocketTransport { Box::pin(async move { #[cfg(not(target_arch = "wasm32"))] { - let socket: TcpWebSocket = connect_native(url, proxy).await?; + let socket = connect_native(url, proxy).await?; Ok(split(Reporting::new(socket))) } @@ -106,7 +106,7 @@ impl WebSocketTransport for DefaultWebsocketTransport { { // The browser dials on our behalf, so a proxy can't be applied here. let _ = proxy; - let socket: WebSocket = WebSocket::connect(url.clone()) + let socket = WebSocket::connect(url.clone()) .await .map_err(Error::transport)?; Ok(split(socket)) @@ -117,7 +117,7 @@ impl WebSocketTransport for DefaultWebsocketTransport { #[cfg(not(target_arch = "wasm32"))] async fn connect_native(url: &Url, proxy: Option) -> Result { - let options: Options = Options::default() + let options = Options::default() .with_limits(MAX_PAYLOAD_READ, MAX_READ_BUFFER) .with_utf8(); @@ -130,7 +130,7 @@ async fn connect_native(url: &Url, proxy: Option) -> Result Date: Thu, 24 Sep 2026 13:03:30 +0200 Subject: [PATCH 7/7] refactor: simplify WebSocket handling --- nostr-sdk/CHANGELOG.md | 21 +-- nostr-sdk/README.md | 15 +- nostr-sdk/examples/local-relay-hyper.rs | 4 +- nostr-sdk/src/error.rs | 3 +- nostr-sdk/src/local_relay/local/inner.rs | 135 +++++++-------- nostr-sdk/src/local_relay/local/mod.rs | 204 +++++++++-------------- nostr-sdk/src/relay/inner.rs | 30 ++-- nostr-sdk/src/transport/websocket.rs | 137 +++++---------- nwc/CHANGELOG.md | 3 +- signer/nostr-connect/CHANGELOG.md | 3 +- 10 files changed, 207 insertions(+), 348 deletions(-) diff --git a/nostr-sdk/CHANGELOG.md b/nostr-sdk/CHANGELOG.md index a3355c1ff..69cfd16bc 100644 --- a/nostr-sdk/CHANGELOG.md +++ b/nostr-sdk/CHANGELOG.md @@ -31,22 +31,19 @@ ### Breaking changes -- Replace `async-wsocket` with `yawc` as the WebSocket implementation. The transport - sink and stream now carry `yawc::frame::Frame` instead of `async_wsocket::Message`, - so custom `WebSocketTransport` implementations build frames with `Frame::text` and - match incoming ones on `Frame::opcode` +- Replace `async-wsocket` with `yawc`. Custom `WebSocketTransport` sinks and streams + now use `yawc::frame::Frame` instead of `async_wsocket::Message`; use `Frame::text` + to build frames and `Frame::opcode` to match them - Remove the `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` - features, which have no equivalent in `yawc`. `ring` and `aws_lc_rs` now select - yawc's rustls backends, and `rustls-tls-webpki-roots` is kept as a no-op because - yawc always trusts the webpki roots. The same features are removed from `nwc` and - `nostr-connect` + features from `nostr-sdk`, `nwc` and `nostr-connect`. `ring` and `aws_lc_rs` now + select yawc's rustls backends; `rustls-tls-webpki-roots` remains a no-op because + yawc always trusts the webpki roots ### Changed -- Route proxied relay connections through `socks5h`, leaving name resolution to the - proxy so `.onion` addresses stay reachable -- Serve the local relay's listener over hyper, which is how yawc reaches its server - side. `LocalRelay::take_connection` is unchanged +- Use `socks5h` for proxied relay connections so the proxy resolves names, including + `.onion` addresses +- Serve the local relay listener with hyper. `LocalRelay::take_connection` is unchanged ### Added diff --git a/nostr-sdk/README.md b/nostr-sdk/README.md index 955b49622..54dd12a2f 100644 --- a/nostr-sdk/README.md +++ b/nostr-sdk/README.md @@ -101,15 +101,12 @@ More examples can be found in the [examples directory](./examples). The following crate feature flags are available: -| Feature | Default | Description | -|---------------------------|:-------:|--------------------------------------------------| -| `ring` | Yes | Enable `ring` crypto provider | -| `rustls-tls-webpki-roots` | Yes | Enable rustls with bundled Mozilla root certs | -| `aws_lc_rs` | No | Enable `aws-lc-rs` crypto provider | -| `native-tls` | No | Enable platform-native TLS | -| `native-tls-vendored` | No | Enable platform-native TLS with vendored OpenSSL | -| `rustls-tls-native-roots` | No | Enable rustls with platform-native root certs | -| `local-relay` | No | Enable `nostr_sdk::local_relay` module | +| Feature | Default | Description | +|---------------------------|:-------:|-----------------------------------------------------| +| `ring` | Yes | Enable the `ring` crypto provider | +| `rustls-tls-webpki-roots` | Yes | Compatibility feature; webpki roots are always used | +| `aws_lc_rs` | No | Enable the `aws-lc-rs` crypto provider | +| `local-relay` | No | Enable the `nostr_sdk::local_relay` module | ## Local Relay supported NIPs diff --git a/nostr-sdk/examples/local-relay-hyper.rs b/nostr-sdk/examples/local-relay-hyper.rs index bd3d1144b..f0f9cf5ed 100644 --- a/nostr-sdk/examples/local-relay-hyper.rs +++ b/nostr-sdk/examples/local-relay-hyper.rs @@ -25,9 +25,7 @@ struct HttpServer { /// Derive the `Sec-WebSocket-Accept` response header from a `Sec-WebSocket-Key` request header. /// -/// This example completes the handshake itself so it can hand the upgraded stream to -/// [`LocalRelay::take_connection`], which is the entry point for servers that already -/// speak HTTP. +/// Complete the handshake before passing the upgraded stream to [`LocalRelay::take_connection`]. pub fn derive_accept_key(request_key: &[u8]) -> String { // ... field is constructed by concatenating /key/ ... // ... with the string "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" (RFC 6455) diff --git a/nostr-sdk/src/error.rs b/nostr-sdk/src/error.rs index d60cfb2d3..d13e3662f 100644 --- a/nostr-sdk/src/error.rs +++ b/nostr-sdk/src/error.rs @@ -7,6 +7,7 @@ use std::error; use nostr::types::RelayUrl; +use yawc::WebSocketError; opaquerr::define_kind! { /// SDK error kind. @@ -53,7 +54,7 @@ opaquerr::define_error! { nostr_gossip::error::Error => ErrorKind::Gossip, faster_hex::Error => ErrorKind::Protocol, negentropy::Error => ErrorKind::Protocol, - yawc::WebSocketError => ErrorKind::Transport, + WebSocketError => ErrorKind::Transport, tokio::sync::oneshot::error::RecvError => ErrorKind::Other, tokio::sync::broadcast::error::RecvError => ErrorKind::Other, #[cfg(any(feature = "local-relay", test))] diff --git a/nostr-sdk/src/local_relay/local/inner.rs b/nostr-sdk/src/local_relay/local/inner.rs index 18f28e6af..74ba823bb 100644 --- a/nostr-sdk/src/local_relay/local/inner.rs +++ b/nostr-sdk/src/local_relay/local/inner.rs @@ -12,6 +12,7 @@ use std::time::Duration; use async_utility::futures_util::stream::SplitSink; use async_utility::futures_util::{SinkExt, StreamExt}; +use hyper::Request; use hyper::body::Incoming; use hyper::server::conn::http1; use hyper::service::service_fn; @@ -323,7 +324,7 @@ impl InnerLocalRelay { let (upgrade_tx, mut upgrade_rx) = mpsc::channel(1); let options = self.websocket_options(); - let service = service_fn(move |mut request: hyper::Request| { + let service = service_fn(move |mut request: Request| { let upgrade_tx = upgrade_tx.clone(); let options = options.clone(); @@ -409,7 +410,7 @@ impl InnerLocalRelay { let payload = msg.payload(); tracing::trace!("Received {}", String::from_utf8_lossy(payload)); let message_size = payload.len(); - match ClientMessage::from_json(payload.as_ref()) { + match ClientMessage::from_json(payload) { Ok(msg) => { self.handle_client_msg( &mut session, @@ -1731,8 +1732,28 @@ where #[cfg(test)] mod tests { + use tokio::io::DuplexStream; + use super::*; + const TEST_ADDR: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); + + struct TestWebSockets { + tx: WsTx, + client: WebSocket, + } + + fn test_websockets() -> TestWebSockets { + let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); + let server = + WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); + let client = + WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); + let (tx, _) = Reporting::new(server).split(); + + TestWebSockets { tx, client } + } + #[derive(Debug, Default)] struct PausedCount { entered: Notify, @@ -1756,7 +1777,7 @@ mod tests { } } - async fn next_frame(socket: &mut WebSocket) -> String { + async fn next_frame(socket: &mut WebSocket) -> String { let frame = tokio::time::timeout(Duration::from_secs(2), socket.next()) .await .unwrap() @@ -2046,22 +2067,16 @@ mod tests { .unwrap(); relay.database.save_event(&event).await.unwrap(); - let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = - WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); - let mut client = - WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); - let (mut server_tx, _) = Reporting::new(server).split(); + let TestWebSockets { mut tx, mut client } = test_websockets(); let mut session = session(None); - let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); let message_size = event.as_json().len() + 10; relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::Event(Cow::Owned(event)), - &addr, + &TEST_ADDR, message_size, ) .await @@ -2086,24 +2101,18 @@ mod tests { .auth_dm(true) .query_policy(ReplaceWithGiftWrap), ); - let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = - WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); - let mut client = - WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); - let (mut server_tx, _) = Reporting::new(server).split(); + let TestWebSockets { mut tx, mut client } = test_websockets(); let mut session = session(None); - let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::Count { subscription_id: Cow::Owned(SubscriptionId::new("gift-wrap")), filter: Cow::Owned(Filter::new().kind(Kind::TextNote)), }, - &addr, + &TEST_ADDR, 0, ) .await @@ -2127,24 +2136,18 @@ mod tests { #[tokio::test] async fn oversized_active_subscription_is_rejected() { let relay = InnerLocalRelay::new(LocalRelayBuilder::default().max_subscription_bytes(10)); - let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = - WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); - let mut client = - WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); - let (mut server_tx, _) = Reporting::new(server).split(); + let TestWebSockets { mut tx, mut client } = test_websockets(); let mut session = session(None); - let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::Req { subscription_id: Cow::Owned(SubscriptionId::new("oversized")), filters: vec![Cow::Owned(Filter::new())], }, - &addr, + &TEST_ADDR, 11, ) .await @@ -2167,24 +2170,18 @@ mod tests { #[tokio::test] async fn excessive_req_filters_are_rejected() { let relay = InnerLocalRelay::new(LocalRelayBuilder::default().max_filters_per_req(1)); - let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = - WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); - let mut client = - WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); - let (mut server_tx, _) = Reporting::new(server).split(); + let TestWebSockets { mut tx, mut client } = test_websockets(); let mut session = session(None); - let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::Req { subscription_id: Cow::Owned(SubscriptionId::new("filters")), filters: vec![Cow::Owned(Filter::new()), Cow::Owned(Filter::new())], }, - &addr, + &TEST_ADDR, 4, ) .await @@ -2205,26 +2202,20 @@ mod tests { #[tokio::test] async fn zero_query_rate_rejects_query_starts() { let relay = InnerLocalRelay::new(LocalRelayBuilder::default().queries_per_minute(0)); - let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = - WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); - let mut client = - WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); - let (mut server_tx, _) = Reporting::new(server).split(); + let TestWebSockets { mut tx, mut client } = test_websockets(); let mut session = session(None); session.query_tokens = Tokens::new(0); session.negentropy_tokens = Tokens::new(0); - let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::Req { subscription_id: Cow::Owned(SubscriptionId::new("limited")), filters: vec![Cow::Owned(Filter::new())], }, - &addr, + &TEST_ADDR, 2, ) .await @@ -2244,12 +2235,12 @@ mod tests { relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::NegMsg { subscription_id: Cow::Owned(SubscriptionId::new("limited-neg-msg")), message: Cow::Borrowed("6100"), }, - &addr, + &TEST_ADDR, 2, ) .await @@ -2269,12 +2260,12 @@ mod tests { relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::Count { subscription_id: Cow::Owned(SubscriptionId::new("limited-count")), filter: Cow::Owned(Filter::new()), }, - &addr, + &TEST_ADDR, 2, ) .await @@ -2294,13 +2285,13 @@ mod tests { relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::NegOpen { subscription_id: Cow::Owned(SubscriptionId::new("limited-neg")), filter: Cow::Owned(Filter::new()), initial_message: Cow::Borrowed(""), }, - &addr, + &TEST_ADDR, 2, ) .await @@ -2321,24 +2312,18 @@ mod tests { #[tokio::test] async fn negentropy_continuations_do_not_consume_query_start_allowance() { let relay = InnerLocalRelay::new(LocalRelayBuilder::default().queries_per_minute(1)); - let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = - WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); - let mut client = - WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); - let (mut server_tx, _) = Reporting::new(server).split(); + let TestWebSockets { mut tx, mut client } = test_websockets(); let mut session = session(None); - let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::NegMsg { subscription_id: Cow::Owned(SubscriptionId::new("neg")), message: Cow::Borrowed("6100"), }, - &addr, + &TEST_ADDR, 2, ) .await @@ -2360,12 +2345,12 @@ mod tests { relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::NegMsg { subscription_id: Cow::Owned(SubscriptionId::new("limited-neg")), message: Cow::Borrowed("6100"), }, - &addr, + &TEST_ADDR, 2, ) .await @@ -2386,12 +2371,12 @@ mod tests { relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::Req { subscription_id: Cow::Owned(SubscriptionId::new("allowed")), filters: vec![Cow::Owned(Filter::new())], }, - &addr, + &TEST_ADDR, 2, ) .await @@ -2409,12 +2394,12 @@ mod tests { relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::Count { subscription_id: Cow::Owned(SubscriptionId::new("limited")), filter: Cow::Owned(Filter::new()), }, - &addr, + &TEST_ADDR, 2, ) .await @@ -2438,22 +2423,16 @@ mod tests { let event = EventBuilder::new(Kind::TextNote, "not an auth event") .finalize(&Keys::generate()) .unwrap(); - let (server_stream, client_stream) = tokio::io::duplex(16 * 1024); - let server = - WebSocket::from_stream(server_stream, Role::Server, Options::default()).unwrap(); - let mut client = - WebSocket::from_stream(client_stream, Role::Client, Options::default()).unwrap(); - let (mut server_tx, _) = Reporting::new(server).split(); + let TestWebSockets { mut tx, mut client } = test_websockets(); let mut session = session(None); session.auth_tokens = Tokens::new(0); - let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); relay .handle_client_msg( &mut session, - &mut server_tx, + &mut tx, ClientMessage::Auth(Cow::Owned(event)), - &addr, + &TEST_ADDR, 0, ) .await diff --git a/nostr-sdk/src/local_relay/local/mod.rs b/nostr-sdk/src/local_relay/local/mod.rs index f16c40c88..25a7a8f8c 100644 --- a/nostr-sdk/src/local_relay/local/mod.rs +++ b/nostr-sdk/src/local_relay/local/mod.rs @@ -177,12 +177,27 @@ mod tests { use nostr::message::{MachineReadablePrefix, RelayMessage}; use nostr::types::Url; use tokio::time; - use yawc::WebSocket; use yawc::frame::{Frame, OpCode}; + use yawc::{TcpWebSocket, WebSocket}; use super::*; use crate::local_relay::{QueryPolicy, QueryPolicyResult}; + async fn connect(relay: &LocalRelay) -> TcpWebSocket { + relay.run().await.unwrap(); + let url = Url::parse(relay.url().await.as_str()).unwrap(); + WebSocket::connect(url).await.unwrap() + } + + async fn next_relay_message(socket: &mut TcpWebSocket) -> RelayMessage<'static> { + let frame = socket + .next() + .await + .expect("WebSocket connection terminated"); + assert_eq!(frame.opcode(), OpCode::Text); + RelayMessage::from_json(frame.payload()).unwrap() + } + #[derive(Debug)] struct RejectQueries; @@ -226,10 +241,8 @@ mod tests { #[tokio::test] async fn test_malformed_client_message_does_not_close_connection() { let relay = LocalRelay::new(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; socket .send(Frame::text( @@ -246,22 +259,15 @@ mod tests { let mut received_notice = false; loop { - let message = socket - .next() - .await - .expect("WebSocket connection terminated"); - - if message.opcode() == OpCode::Text { - match RelayMessage::from_json(message.payload()).unwrap() { - RelayMessage::Notice(..) => received_notice = true, - RelayMessage::EndOfStoredEvents(subscription_id) - if subscription_id.as_str() == "valid" => - { - assert!(received_notice); - break; - } - _ => {} + match next_relay_message(&mut socket).await { + RelayMessage::Notice(..) => received_notice = true, + RelayMessage::EndOfStoredEvents(subscription_id) + if subscription_id.as_str() == "valid" => + { + assert!(received_notice); + break; } + _ => {} } } }) @@ -272,10 +278,8 @@ mod tests { #[tokio::test] async fn test_malformed_negentropy_messages_do_not_close_connection() { let relay = LocalRelay::new(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; socket .send(Frame::text(r#"["NEG-OPEN","neg-odd",{},"abc"]"#)) @@ -298,25 +302,18 @@ mod tests { let mut neg_errors: usize = 0; loop { - let message = socket - .next() - .await - .expect("WebSocket connection terminated"); - - if message.opcode() == OpCode::Text { - match RelayMessage::from_json(message.payload()).unwrap() { - RelayMessage::NegErr { message, .. } => { - assert_eq!(message, "error: invalid negentropy message"); - neg_errors += 1; - } - RelayMessage::EndOfStoredEvents(subscription_id) - if subscription_id.as_str() == "valid" => - { - assert_eq!(neg_errors, 3); - break; - } - _ => {} + match next_relay_message(&mut socket).await { + RelayMessage::NegErr { message, .. } => { + assert_eq!(message, "error: invalid negentropy message"); + neg_errors += 1; } + RelayMessage::EndOfStoredEvents(subscription_id) + if subscription_id.as_str() == "valid" => + { + assert_eq!(neg_errors, 3); + break; + } + _ => {} } } }) @@ -327,10 +324,8 @@ mod tests { #[tokio::test] async fn test_invalid_neg_msg_terminates_only_the_negentropy_subscription() { let relay = LocalRelay::new(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; // Open a valid negentropy subscription let mut storage = NegentropyStorageVector::new(); @@ -344,12 +339,8 @@ mod tests { .await .unwrap(); - let reply = socket.next().await.unwrap(); - assert_eq!(reply.opcode(), OpCode::Text); - assert!(matches!( - RelayMessage::from_json(reply.payload()).unwrap(), - RelayMessage::NegMsg { .. } - )); + let reply = next_relay_message(&mut socket).await; + assert!(matches!(reply, RelayMessage::NegMsg { .. })); // A malformed payload must terminate only this negentropy subscription socket @@ -357,10 +348,9 @@ mod tests { .await .unwrap(); - let neg_err = socket.next().await.unwrap(); - assert_eq!(neg_err.opcode(), OpCode::Text); + let neg_err = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.payload()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -374,10 +364,9 @@ mod tests { .await .unwrap(); - let neg_err = socket.next().await.unwrap(); - assert_eq!(neg_err.opcode(), OpCode::Text); + let neg_err = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.payload()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -390,10 +379,9 @@ mod tests { .await .unwrap(); - let eose = socket.next().await.unwrap(); - assert_eq!(eose.opcode(), OpCode::Text); + let eose = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(eose.payload()).unwrap(), + eose, RelayMessage::EndOfStoredEvents(subscription_id) if subscription_id.as_str() == "valid" )); @@ -404,26 +392,19 @@ mod tests { let relay = LocalRelay::builder() .nip42(crate::local_relay::LocalRelayBuilderNip42::read()) .build(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; socket .send(Frame::text(r#"["COUNT","count",{}]"#)) .await .unwrap(); - let auth = socket.next().await.unwrap(); - assert_eq!(auth.opcode(), OpCode::Text); - assert!(matches!( - RelayMessage::from_json(auth.payload()).unwrap(), - RelayMessage::Auth { .. } - )); + let auth = next_relay_message(&mut socket).await; + assert!(matches!(auth, RelayMessage::Auth { .. })); - let closed = socket.next().await.unwrap(); - assert_eq!(closed.opcode(), OpCode::Text); + let closed = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(closed.payload()).unwrap(), + closed, RelayMessage::Closed { subscription_id, message, @@ -437,26 +418,19 @@ mod tests { let relay = LocalRelay::builder() .nip42(crate::local_relay::LocalRelayBuilderNip42::read()) .build(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; socket .send(Frame::text(r#"["NEG-OPEN","neg",{},""]"#)) .await .unwrap(); - let auth = socket.next().await.unwrap(); - assert_eq!(auth.opcode(), OpCode::Text); - assert!(matches!( - RelayMessage::from_json(auth.payload()).unwrap(), - RelayMessage::Auth { .. } - )); + let auth = next_relay_message(&mut socket).await; + assert!(matches!(auth, RelayMessage::Auth { .. })); - let neg_err = socket.next().await.unwrap(); - assert_eq!(neg_err.opcode(), OpCode::Text); + let neg_err = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.payload()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -468,19 +442,16 @@ mod tests { #[tokio::test] async fn test_query_policy_is_applied_to_count_and_negentropy() { let relay = LocalRelay::builder().query_policy(RejectQueries).build(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; socket .send(Frame::text(r#"["COUNT","count",{}]"#)) .await .unwrap(); - let closed = socket.next().await.unwrap(); - assert_eq!(closed.opcode(), OpCode::Text); + let closed = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(closed.payload()).unwrap(), + closed, RelayMessage::Closed { subscription_id, message, @@ -493,10 +464,9 @@ mod tests { .await .unwrap(); - let neg_err = socket.next().await.unwrap(); - assert_eq!(neg_err.opcode(), OpCode::Text); + let neg_err = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.payload()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -510,19 +480,16 @@ mod tests { let relay = LocalRelay::builder() .max_negentropy_subscriptions(0) .build(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; socket .send(Frame::text(r#"["NEG-OPEN","neg",{},""]"#)) .await .unwrap(); - let neg_err = socket.next().await.unwrap(); - assert_eq!(neg_err.opcode(), OpCode::Text); + let neg_err = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.payload()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -534,19 +501,16 @@ mod tests { #[tokio::test] async fn test_subscription_id_limit_counts_utf8_bytes() { let relay = LocalRelay::builder().max_subid_length(3).build(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; socket .send(Frame::text(r#"["NEG-OPEN","éé",{},""]"#)) .await .unwrap(); - let neg_err = socket.next().await.unwrap(); - assert_eq!(neg_err.opcode(), OpCode::Text); + let neg_err = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.payload()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -569,19 +533,16 @@ mod tests { .await .unwrap(); } - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; socket .send(Frame::text(r#"["NEG-OPEN","neg",{},""]"#)) .await .unwrap(); - let neg_err = socket.next().await.unwrap(); - assert_eq!(neg_err.opcode(), OpCode::Text); + let neg_err = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.payload()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -593,10 +554,8 @@ mod tests { #[tokio::test] async fn test_default_message_limit_allows_301_protocol_frames() { let relay = LocalRelay::new(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; time::timeout(Duration::from_secs(5), async { for _ in 0..301 { @@ -610,10 +569,9 @@ mod tests { .await .unwrap(); - let eose = socket.next().await.unwrap(); - assert_eq!(eose.opcode(), OpCode::Text); + let eose = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(eose.payload()).unwrap(), + eose, RelayMessage::EndOfStoredEvents(subscription_id) if subscription_id.as_str() == "valid" )); @@ -625,10 +583,8 @@ mod tests { #[tokio::test] async fn test_default_message_limit_closes_rapid_burst_over_6000_frames() { let relay = LocalRelay::new(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; time::timeout(Duration::from_secs(5), async { for _ in 0..6_001 { @@ -648,20 +604,14 @@ mod tests { #[tokio::test] async fn test_configured_message_limit_is_exact_for_binary_frames() { let relay = LocalRelay::builder().messages_per_minute(3).build(); - relay.run().await.unwrap(); - let url = Url::parse(relay.url().await.as_str()).unwrap(); - let mut socket = WebSocket::connect(url).await.unwrap(); + let mut socket = connect(&relay).await; time::timeout(Duration::from_secs(5), async { for byte in 1..=3 { socket.send(Frame::binary(vec![byte])).await.unwrap(); - let notice = socket.next().await.unwrap(); - assert_eq!(notice.opcode(), OpCode::Text); - assert!(matches!( - RelayMessage::from_json(notice.payload()).unwrap(), - RelayMessage::Notice(..) - )); + let notice = next_relay_message(&mut socket).await; + assert!(matches!(notice, RelayMessage::Notice(..))); } socket.send(Frame::binary(vec![4])).await.unwrap(); diff --git a/nostr-sdk/src/relay/inner.rs b/nostr-sdk/src/relay/inner.rs index 28404c1fa..e942722a5 100644 --- a/nostr-sdk/src/relay/inner.rs +++ b/nostr-sdk/src/relay/inner.rs @@ -884,7 +884,7 @@ impl InnerRelay { tracing::debug!("Sending '{json}' to '{}' (size: {size} bytes)", self.url); // Compose WebSocket text messages - let msg: Frame = Frame::text(json); + let msg = Frame::text(json); // Send WebSocket messages send_ws_msg(ws_tx, msg).await?; @@ -945,18 +945,14 @@ impl InnerRelay { let _ping = ping; while let Some(frame) = ws_rx.next().await { - let frame: Frame = frame?; + let frame = frame?; match frame.opcode() { - // The transport asks yawc to validate UTF-8, so an invalid payload closes the - // connection before it reaches here. Handle it anyway rather than panicking on - // `Frame::as_str`. - OpCode::Text => match str::from_utf8(frame.payload()) { - Ok(json) => self.handle_relay_message(json, &ingester_tx).await, - Err(..) => { - return Err(Error::protocol_msg("relay message is not valid UTF-8")); - } - }, + OpCode::Text => { + let json = str::from_utf8(frame.payload()) + .map_err(|_| Error::protocol_msg("relay message is not valid UTF-8"))?; + self.handle_relay_message(json, &ingester_tx).await; + } OpCode::Binary => { tracing::warn!(url = %self.url, "Binary messages aren't supported."); } @@ -966,28 +962,22 @@ impl InnerRelay { return Err(Error::protocol_msg("can't parse pong")); }; - // Nonce from big-endian bytes - let nonce: u64 = u64::from_be_bytes(nonce); - - // Get last nonce - let last_nonce: u64 = ping.last_nonce(); + let nonce = u64::from_be_bytes(nonce); + let last_nonce = ping.last_nonce(); - // Check if last nonce not matches the received one if last_nonce != nonce { return Err(Error::pong_not_match(last_nonce, nonce)); } - // Set ping as replied ping.set_replied(true); - // Save latency let sent_at = ping.sent_at().await; self.stats.save_latency(sent_at.elapsed()); } #[cfg(not(target_arch = "wasm32"))] OpCode::Close => { if let Some(code) = frame.close_code() { - let reason: &str = frame.close_reason().ok().flatten().unwrap_or_default(); + let reason = frame.close_reason().ok().flatten().unwrap_or(""); tracing::info!(?code, reason, "Connection closed by peer."); } break; diff --git a/nostr-sdk/src/transport/websocket.rs b/nostr-sdk/src/transport/websocket.rs index b1e5057ed..ded588a83 100644 --- a/nostr-sdk/src/transport/websocket.rs +++ b/nostr-sdk/src/transport/websocket.rs @@ -10,7 +10,6 @@ use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; -use futures::stream::SplitSink; use futures::{Sink, SinkExt, Stream, StreamExt, TryStreamExt}; use nostr::types::Url; #[cfg(target_arch = "wasm32")] @@ -18,7 +17,7 @@ use yawc::WebSocket; use yawc::WebSocketError; use yawc::frame::Frame; #[cfg(not(target_arch = "wasm32"))] -use yawc::{HttpRequest, Options, Proxy, TcpWebSocket, WebSocket}; +use yawc::{HttpRequest, Options, Proxy, WebSocket}; use crate::error::Error; use crate::future::BoxedFuture; @@ -98,46 +97,34 @@ impl WebSocketTransport for DefaultWebsocketTransport { Box::pin(async move { #[cfg(not(target_arch = "wasm32"))] { - let socket = connect_native(url, proxy).await?; - Ok(split(Reporting::new(socket))) + let mut connection = WebSocket::connect(url.clone()) + .with_options( + Options::default() + .with_limits(MAX_PAYLOAD_READ, MAX_READ_BUFFER) + .with_utf8(), + ) + .with_request(HttpRequest::builder().header("user-agent", USER_AGENT)); + + if let Some(proxy) = proxy { + // Resolve names at the proxy so `.onion` addresses work. + let url = + Url::parse(&format!("socks5h://{proxy}")).map_err(Error::transport)?; + connection = connection.with_proxy(Proxy::socks5(url)?); + } + + Ok(split(Reporting::new(connection.await?))) } #[cfg(target_arch = "wasm32")] { // The browser dials on our behalf, so a proxy can't be applied here. let _ = proxy; - let socket = WebSocket::connect(url.clone()) - .await - .map_err(Error::transport)?; - Ok(split(socket)) + Ok(split(WebSocket::connect(url.clone()).await?)) } }) } } -#[cfg(not(target_arch = "wasm32"))] -async fn connect_native(url: &Url, proxy: Option) -> Result { - let options = Options::default() - .with_limits(MAX_PAYLOAD_READ, MAX_READ_BUFFER) - .with_utf8(); - - let request = HttpRequest::builder().header("user-agent", USER_AGENT); - - let mut builder = WebSocket::connect(url.clone()) - .with_options(options) - .with_request(request); - - if let Some(proxy) = proxy { - // `socks5h` leaves name resolution to the proxy, which is what makes `.onion` - // addresses resolvable at all. - let url = Url::parse(&format!("socks5h://{proxy}")).map_err(Error::transport)?; - builder = builder.with_proxy(Proxy::socks5(url).map_err(Error::transport)?); - } - - builder.await.map_err(Error::transport) -} - -/// Split a socket into the boxed sink and stream halves the transport hands back. fn split(socket: T) -> (WebSocketSink, WebSocketStream) where T: Sink @@ -150,21 +137,16 @@ where // NOTE: don't use sink_map_err here, as it may cause panics! // Issue: https://github.com/nostrdevkit/nostr/issues/984 - let sink: WebSocketSink = Box::pin(TransportSink(tx)); - let stream: WebSocketStream = Box::pin(rx.map_err(Error::transport)); - - (sink, stream) + ( + Box::pin(TransportSink(tx)), + Box::pin(rx.map_err(Error::transport)), + ) } -/// A socket that reports the read errors yawc's own [`Stream`] impl hides. -/// -/// `impl Stream for WebSocket` maps a failure to the end of the stream, which would make a -/// broken connection indistinguishable from one the relay closed cleanly. Polling the frame -/// directly keeps the two apart, at the cost of tracking the terminal state here. +/// Reports read errors that yawc's [`Stream`] implementation hides. #[cfg(not(target_arch = "wasm32"))] pub(crate) struct Reporting { socket: WebSocket, - /// Set once a read has failed, so the socket is not polled again afterwards. failed: bool, } @@ -192,13 +174,9 @@ where return Poll::Ready(None); } - match futures::ready!(this.socket.poll_next_frame(cx)) { - Ok(frame) => Poll::Ready(Some(Ok(frame))), - Err(e) => { - this.failed = true; - Poll::Ready(Some(Err(e))) - } - } + let result = futures::ready!(this.socket.poll_next_frame(cx)); + this.failed = result.is_err(); + Poll::Ready(Some(result)) } } @@ -210,54 +188,44 @@ where type Error = WebSocketError; fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.socket).poll_ready(cx) + self.socket.poll_ready_unpin(cx) } fn start_send(mut self: Pin<&mut Self>, item: Frame) -> Result<(), Self::Error> { - Pin::new(&mut self.socket).start_send(item) + self.socket.start_send_unpin(item) } fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.socket).poll_flush(cx) + self.socket.poll_flush_unpin(cx) } fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.socket).poll_close(cx) + self.socket.poll_close_unpin(cx) } } -struct TransportSink(SplitSink) -where - T: Sink; +struct TransportSink(S); -impl Sink for TransportSink +impl Sink for TransportSink where - T: Sink + Unpin, + S: Sink + Unpin, { type Error = Error; fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.0) - .poll_ready_unpin(cx) - .map_err(Error::transport) + self.0.poll_ready_unpin(cx).map_err(Error::from) } fn start_send(mut self: Pin<&mut Self>, item: Frame) -> Result<(), Self::Error> { - Pin::new(&mut self.0) - .start_send_unpin(item) - .map_err(Error::transport) + self.0.start_send_unpin(item).map_err(Error::from) } fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.0) - .poll_flush_unpin(cx) - .map_err(Error::transport) + self.0.poll_flush_unpin(cx).map_err(Error::from) } fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.0) - .poll_close_unpin(cx) - .map_err(Error::transport) + self.0.poll_close_unpin(cx).map_err(Error::from) } } @@ -291,30 +259,6 @@ mod tests { .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) } - #[tokio::test] - async fn default_transport_sends_user_agent() -> Result<(), Box> { - let listener = TcpListener::bind("127.0.0.1:0").await?; - let address = listener.local_addr()?; - - let server = tokio::spawn(async move { - let (mut stream, _) = listener.accept().await?; - read_request(&mut stream).await - }); - - let url = Url::parse(&format!("ws://{address}"))?; - assert!(DefaultWebsocketTransport.connect(&url, None).await.is_err()); - - let request = server.await??; - let user_agent = request.lines().find_map(|line| { - let (name, value) = line.split_once(':')?; - name.eq_ignore_ascii_case("user-agent") - .then(|| value.trim()) - }); - assert_eq!(user_agent, Some(USER_AGENT)); - - Ok(()) - } - /// Serve one SOCKS5 CONNECT, then read whatever the client tunnels through it. /// /// Nothing answers the WebSocket handshake, so the client's connect fails. What the test @@ -351,7 +295,7 @@ mod tests { let domain = String::from_utf8(domain).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; - let target: String = format!("{domain}:{}", u16::from_be_bytes(port)); + let target = format!("{domain}:{}", u16::from_be_bytes(port)); // Success, with a bound address of 0.0.0.0:0. stream @@ -381,6 +325,11 @@ mod tests { let (target, request) = server.await??; assert_eq!(target, "relay.invalid:8080"); assert!(request.starts_with("GET / HTTP/1.1"), "{request}"); + assert!( + request + .lines() + .any(|line| line.eq_ignore_ascii_case(&format!("user-agent: {USER_AGENT}"))) + ); Ok(()) } diff --git a/nwc/CHANGELOG.md b/nwc/CHANGELOG.md index ebbddd38c..5a8d3d390 100644 --- a/nwc/CHANGELOG.md +++ b/nwc/CHANGELOG.md @@ -31,8 +31,7 @@ ### Breaking changes -- Remove the `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` features, - following their removal from `nostr-sdk` +- Remove the `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` features ## v0.45.1 - 2026/09/11 diff --git a/signer/nostr-connect/CHANGELOG.md b/signer/nostr-connect/CHANGELOG.md index e93b90b66..6fed3e45d 100644 --- a/signer/nostr-connect/CHANGELOG.md +++ b/signer/nostr-connect/CHANGELOG.md @@ -31,8 +31,7 @@ ### Breaking changes -- Remove the `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` features, - following their removal from `nostr-sdk` +- Remove the `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` features ## v0.45.2 - 2026/09/11