diff --git a/Cargo.lock b/Cargo.lock index fd14a460f..2af5492ac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -124,26 +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-rustls", - "tokio-socks", - "tokio-tungstenite", - "url", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "atomic-destructor" version = "0.3.0" @@ -360,7 +340,7 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" dependencies = [ - "nom", + "nom 7.1.3", ] [[package]] @@ -616,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" @@ -783,6 +757,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 +792,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 +1160,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 +1194,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots 1.0.6", + "webpki-roots", ] [[package]] @@ -1448,7 +1408,7 @@ dependencies = [ "jni-sys", "log", "simd_cesu8", - "thiserror 2.0.18", + "thiserror", "walkdir", "windows-link", ] @@ -1707,23 +1667,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 +1689,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.5" @@ -1936,7 +1888,6 @@ name = "nostr-sdk" version = "0.45.4" dependencies = [ "async-utility", - "async-wsocket", "base64", "bitcoin_hashes 1.1.0", "faster-hex", @@ -1960,6 +1911,7 @@ dependencies = [ "tracing", "tracing-subscriber", "universal-time", + "yawc", ] [[package]] @@ -1987,7 +1939,7 @@ dependencies = [ "futures", "libc", "libsodium-sys-stable", - "thiserror 2.0.18", + "thiserror", "tokio", "tracing", ] @@ -2063,59 +2015,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" @@ -2207,6 +2106,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" @@ -2287,7 +2206,7 @@ dependencies = [ "rustc-hash 2.1.1", "rustls", "socket2", - "thiserror 2.0.18", + "thiserror", "tokio", "tracing", "web-time", @@ -2308,7 +2227,7 @@ dependencies = [ "rustls", "rustls-pki-types", "slab", - "thiserror 2.0.18", + "thiserror", "tinyvec", "tracing", "web-time", @@ -2507,7 +2426,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", - "webpki-roots 1.0.6", + "webpki-roots", ] [[package]] @@ -2537,7 +2456,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a8a1f2315036ef6b1fbacd1972e8ee7688030b0a2121edfc2a6550febd41574d" dependencies = [ "hashbrown 0.16.1", - "thiserror 2.0.18", + "thiserror", ] [[package]] @@ -2608,6 +2527,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4" dependencies = [ "aws-lc-rs", + "log", "once_cell", "ring", "rustls-pki-types", @@ -2616,18 +2536,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" @@ -2681,15 +2589,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" @@ -2728,35 +2627,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" @@ -3018,33 +2900,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]] @@ -3109,15 +2971,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" @@ -3129,16 +2982,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" @@ -3149,18 +2992,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" @@ -3173,25 +3004,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", - "native-tls", - "rustls", - "rustls-native-certs", - "rustls-pki-types", - "tokio", - "tokio-native-tls", - "tokio-rustls", - "tungstenite", - "webpki-roots 0.26.11", -] - [[package]] name = "tokio-util" version = "0.7.18" @@ -3317,26 +3129,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", - "native-tls", - "rand 0.9.2", - "rustls", - "rustls-pki-types", - "sha1", - "thiserror 2.0.18", - "utf-8", -] - [[package]] name = "typed-path" version = "0.12.3" @@ -3436,12 +3228,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" @@ -3662,15 +3448,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" @@ -4004,6 +3781,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/CHANGELOG.md b/nostr-sdk/CHANGELOG.md index 871813789..69cfd16bc 100644 --- a/nostr-sdk/CHANGELOG.md +++ b/nostr-sdk/CHANGELOG.md @@ -29,6 +29,22 @@ ## Unreleased +### Breaking changes + +- 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 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 + +- 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 - Add `LocalRelay::connections_left` (https://github.com/nostrdevkit/nostr/pull/1459) diff --git a/nostr-sdk/Cargo.toml b/nostr-sdk/Cargo.toml index 1a1dfaebe..22ca5e1ae 100644 --- a/nostr-sdk/Cargo.toml +++ b/nostr-sdk/Cargo.toml @@ -17,19 +17,18 @@ 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"] -local-relay = ["dep:nostr-memory", "tokio/net"] +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:hyper", "dep:hyper-util", "dep:nostr-memory", "tokio/net"] [dependencies] async-utility.workspace = true -async-wsocket = { version = "0.17", features = ["socks"] } 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"] } @@ -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/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 3463df46c..f0f9cf5ed 100644 --- a/nostr-sdk/examples/local-relay-hyper.rs +++ b/nostr-sdk/examples/local-relay-hyper.rs @@ -23,11 +23,9 @@ 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]. +/// 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 03bb4ba14..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,8 +54,7 @@ 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, + 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 85cf059b3..74ba823bb 100644 --- a/nostr-sdk/src/local_relay/local/inner.rs +++ b/nostr-sdk/src/local_relay/local/inner.rs @@ -12,8 +12,11 @@ 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::Request; +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 +24,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 +38,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,13 +295,11 @@ impl InnerLocalRelay { tokio::time::sleep(unresponsive_connection).await; } - let ws_stream = - WebSocketStream::from_raw_socket(stream, Role::Server, Some(self.websocket_config())) - .await; - - self.handle_websocket(ws_stream, addr, permit).await?; + let socket = WebSocket::from_stream(stream, Role::Server, self.websocket_options()) + .map_err(Error::transport)?; - Ok(()) + self.handle_websocket(Reporting::new(socket), addr, permit) + .await } /// Pass bare [TcpStream] for handling @@ -306,39 +310,65 @@ impl InnerLocalRelay { handshake_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; } - // Take the connection permit before doing the handshake - let conn_permit: OwnedSemaphorePermit = - self.connections_limit.clone().try_acquire_owned()?; + // Take the connection permit before doing the handshake. + let connection_permit = self.connections_limit.clone().try_acquire_owned()?; + + // 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: Request| { + let upgrade_tx = upgrade_tx.clone(); + let options = options.clone(); + + async move { + let (response, upgrade) = WebSocket::upgrade_with_options(&mut request, options)?; + let _ = upgrade_tx.send(upgrade).await; + + 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)?; + + upgrade_rx + .recv() + .await + .ok_or_else(|| { + Error::with_static_message(ErrorKind::Transport, "WebSocket upgrade failed") + })? + .await + .map_err(Error::transport) + }) .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(handshake_permit); - self.handle_websocket(ws_stream, addr, conn_permit).await?; - - Ok(()) + self.handle_websocket(Reporting::new(socket), addr, connection_permit) + .await } /// Handle websocket connection async fn handle_websocket( &self, - ws_stream: WebSocketStream, + ws_stream: Reporting, addr: SocketAddr, _permit: OwnedSemaphorePermit, ) -> Result<(), Error> @@ -375,11 +405,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) { Ok(msg) => { self.handle_client_msg( &mut session, @@ -402,17 +433,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}"), @@ -476,10 +507,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( @@ -1514,10 +1548,9 @@ 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(()) + .map_err(Error::other) } async fn send_query_rate_limit_error( @@ -1699,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, @@ -1724,15 +1777,12 @@ mod tests { } } - async fn next_frame(socket: &mut WebSocketStream) -> String { - tokio::time::timeout(Duration::from_secs(2), socket.next()) + async fn next_frame(socket: &mut WebSocket) -> String { + let frame = tokio::time::timeout(Duration::from_secs(2), socket.next()) .await .unwrap() - .unwrap() - .unwrap() - .into_text() - .unwrap() - .to_string() + .unwrap(); + String::from_utf8(frame.payload().to_vec()).unwrap() } #[tokio::test] @@ -1757,12 +1807,10 @@ mod tests { .handle_upgraded_connection(server, "127.0.0.1:1234".parse().unwrap()) .await }); - let mut client = WebSocketStream::from_raw_socket(client, Role::Client, None).await; + let mut client = WebSocket::from_stream(client, Role::Client, Options::default()).unwrap(); for id in ["first", "second"] { client - .send(Message::Text( - format!(r#"["REQ","{id}",{{"kinds":[1]}}]"#).into(), - )) + .send(Frame::text(format!(r#"["REQ","{id}",{{"kinds":[1]}}]"#))) .await .unwrap(); assert_eq!(next_frame(&mut client).await, format!(r#"["EOSE","{id}"]"#)); @@ -1776,18 +1824,17 @@ mod tests { .handle_upgraded_connection(server, "127.0.0.1:1235".parse().unwrap()) .await }); - let mut healthy = WebSocketStream::from_raw_socket(healthy, Role::Client, None).await; + let mut healthy = + WebSocket::from_stream(healthy, Role::Client, Options::default()).unwrap(); healthy - .send(Message::Text(r#"["REQ","healthy",{"kinds":[1]}]"#.into())) + .send(Frame::text(r#"["REQ","healthy",{"kinds":[1]}]"#)) .await .unwrap(); assert_eq!(next_frame(&mut healthy).await, r#"["EOSE","healthy"]"#); // Hold one session inside a query while filling its live broadcast buffer. client - .send(Message::Text( - r#"["COUNT","pause",{"kinds":[1],"limit":42}]"#.into(), - )) + .send(Frame::text(r#"["COUNT","pause",{"kinds":[1],"limit":42}]"#)) .await .unwrap(); tokio::time::timeout(Duration::from_secs(2), policy.entered.notified()) @@ -1857,7 +1904,7 @@ mod tests { ); for id in ["first", "second"] { client - .send(Message::Text(format!(r#"["CLOSE","{id}"]"#).into())) + .send(Frame::text(format!(r#"["CLOSE","{id}"]"#))) .await .unwrap(); } @@ -1865,7 +1912,7 @@ mod tests { // Reusing the connection must not retain the old subscriptions' byte budget. client - .send(Message::Text(r#"["REQ","recovered",{"kinds":[1]}]"#.into())) + .send(Frame::text(r#"["REQ","recovered",{"kinds":[1]}]"#)) .await .unwrap(); let frame = next_frame(&mut client).await; @@ -1960,7 +2007,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!( @@ -1974,8 +2021,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); } @@ -1987,12 +2034,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); } @@ -2020,31 +2067,25 @@ mod tests { .unwrap(); 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 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 .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, @@ -2060,78 +2101,62 @@ mod tests { .auth_dm(true) .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 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 .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:") )); } #[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 = 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 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 .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, @@ -2145,33 +2170,27 @@ 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 = 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 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 .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, @@ -2183,35 +2202,29 @@ 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 = 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 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 .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, @@ -2222,23 +2235,21 @@ 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 .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, @@ -2249,23 +2260,21 @@ 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 .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, @@ -2276,24 +2285,22 @@ 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 .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, @@ -2305,33 +2312,27 @@ 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 = 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 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 .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, @@ -2344,23 +2345,21 @@ 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 .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, @@ -2372,23 +2371,21 @@ 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 .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" )); @@ -2397,23 +2394,21 @@ 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 .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, @@ -2428,34 +2423,27 @@ 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 = 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 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 .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 3d000e368..25a7a8f8c 100644 --- a/nostr-sdk/src/local_relay/local/mod.rs +++ b/nostr-sdk/src/local_relay/local/mod.rs @@ -169,18 +169,35 @@ 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::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; @@ -224,21 +241,17 @@ 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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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(); @@ -246,23 +259,15 @@ mod tests { let mut received_notice = false; loop { - let message = socket - .next() - .await - .expect("WebSocket connection terminated") - .unwrap(); - - if let Message::Text(json) = message { - match RelayMessage::from_json(json.as_bytes()).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; } + _ => {} } } }) @@ -273,31 +278,23 @@ 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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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(); @@ -305,26 +302,18 @@ mod tests { let mut neg_errors: usize = 0; loop { - let message = socket - .next() - .await - .expect("WebSocket connection terminated") - .unwrap(); - - if let Message::Text(json) = message { - match RelayMessage::from_json(json.as_bytes()).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; } + _ => {} } } }) @@ -335,12 +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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; // Open a valid negentropy subscription let mut storage = NegentropyStorageVector::new(); @@ -348,33 +333,24 @@ 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"); - }; - assert!(matches!( - RelayMessage::from_json(reply.as_bytes()).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 - .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 = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -384,16 +360,13 @@ 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 = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -402,16 +375,13 @@ 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 = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(eose.as_bytes()).unwrap(), + eose, RelayMessage::EndOfStoredEvents(subscription_id) if subscription_id.as_str() == "valid" )); @@ -422,32 +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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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"); - }; - assert!(matches!( - RelayMessage::from_json(auth.as_bytes()).unwrap(), - RelayMessage::Auth { .. } - )); + let auth = next_relay_message(&mut socket).await; + assert!(matches!(auth, RelayMessage::Auth { .. })); - let closed = socket.next().await.unwrap().unwrap(); - let Message::Text(closed) = closed else { - panic!("unexpected websocket message"); - }; + let closed = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(closed.as_bytes()).unwrap(), + closed, RelayMessage::Closed { subscription_id, message, @@ -461,32 +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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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"); - }; - assert!(matches!( - RelayMessage::from_json(auth.as_bytes()).unwrap(), - RelayMessage::Auth { .. } - )); + let auth = next_relay_message(&mut socket).await; + assert!(matches!(auth, 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 = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -498,23 +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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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 = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(closed.as_bytes()).unwrap(), + closed, RelayMessage::Closed { subscription_id, message, @@ -523,16 +460,13 @@ 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 = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -546,23 +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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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 = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -574,23 +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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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 = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -613,23 +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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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 = next_relay_message(&mut socket).await; assert!(matches!( - RelayMessage::from_json(neg_err.as_bytes()).unwrap(), + neg_err, RelayMessage::NegErr { subscription_id, message, @@ -641,34 +554,26 @@ 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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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 = next_relay_message(&mut socket).await; 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::EndOfStoredEvents(subscription_id) + if subscription_id.as_str() == "valid" )); }) .await @@ -678,26 +583,19 @@ 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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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"); @@ -706,33 +604,19 @@ 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, &ConnectionMode::direct()) - .await - .unwrap(); + let mut socket = connect(&relay).await; 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(); - assert!(matches!( - notice, - Message::Text(json) - if matches!( - RelayMessage::from_json(json.as_bytes()).unwrap(), - RelayMessage::Notice(..) - ) - )); + socket.send(Frame::binary(vec![byte])).await.unwrap(); + let notice = next_relay_message(&mut socket).await; + assert!(matches!(notice, 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/relay/inner.rs b/nostr-sdk/src/relay/inner.rs index 7a716b0bc..e942722a5 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::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,44 @@ 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?; + + match frame.opcode() { + 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."); } #[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); + let nonce = u64::from_be_bytes(nonce); + let last_nonce = ping.last_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")); - } + if last_nonce != nonce { + return Err(Error::pong_not_match(last_nonce, nonce)); } + + ping.set_replied(true); + + 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 = frame.close_reason().ok().flatten().unwrap_or(""); + tracing::info!(?code, reason, "Connection closed by peer."); + } break; } - #[cfg(not(target_arch = "wasm32"))] _ => {} } } @@ -1857,7 +1854,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..ded588a83 100644 --- a/nostr-sdk/src/transport/websocket.rs +++ b/nostr-sdk/src/transport/websocket.rs @@ -10,12 +10,14 @@ 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, WebSocket}; use crate::error::Error; use crate::future::BoxedFuture; @@ -23,10 +25,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,105 +95,242 @@ 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 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")] - let connection = WebSocket::connect(url, &mode); + { + // The browser dials on our behalf, so a proxy can't be applied here. + let _ = proxy; + Ok(split(WebSocket::connect(url.clone()).await?)) + } + }) + } +} - let socket: WebSocket = connection.await.map_err(Error::transport)?; +fn split(socket: T) -> (WebSocketSink, WebSocketStream) +where + T: Sink + + Stream> + + Send + + Unpin + + 'static, +{ + 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 + ( + Box::pin(TransportSink(tx)), + Box::pin(rx.map_err(Error::transport)), + ) +} - // Split sink and stream - let (tx, rx) = socket.split(); +/// Reports read errors that yawc's [`Stream`] implementation hides. +#[cfg(not(target_arch = "wasm32"))] +pub(crate) struct Reporting { + socket: WebSocket, + failed: bool, +} - // 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; +#[cfg(not(target_arch = "wasm32"))] +impl Reporting { + pub(crate) fn new(socket: WebSocket) -> Self { + Self { + socket, + failed: false, + } + } +} - Ok((sink, stream)) - }) +#[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(); + + if this.failed { + return Poll::Ready(None); + } + + let result = futures::ready!(this.socket.poll_next_frame(cx)); + this.failed = result.is_err(); + Poll::Ready(Some(result)) } } -struct TransportSink(SplitSink); +#[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> { + self.socket.poll_ready_unpin(cx) + } + + fn start_send(mut self: Pin<&mut Self>, item: Frame) -> Result<(), Self::Error> { + self.socket.start_send_unpin(item) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.socket.poll_flush_unpin(cx) + } -impl Sink for TransportSink { + fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.socket.poll_close_unpin(cx) + } +} + +struct TransportSink(S); + +impl Sink for TransportSink +where + 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: Message) -> Result<(), Self::Error> { - Pin::new(&mut self.0) - .start_send_unpin(item) - .map_err(Error::transport) + fn start_send(mut self: Pin<&mut Self>, item: Frame) -> Result<(), Self::Error> { + 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) } } #[cfg(all(test, not(target_arch = "wasm32")))] mod tests { - use tokio::io::AsyncReadExt; + use std::error::Error as StdError; + use std::io; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; use super::*; - #[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(); - - 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; + /// Read the HTTP request a client wrote, up to the blank line that ends its head. + async fn read_request(stream: &mut S) -> io::Result + 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?; + 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)) + } - let url = Url::parse(&format!("ws://{address}")).unwrap(); - assert!(DefaultWebsocketTransport.connect(&url, None).await.is_err()); + /// 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) -> 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?; + assert_eq!(greeting[0], 0x05); + let mut methods = vec![0u8; greeting[1] as usize]; + stream.read_exact(&mut methods).await?; + + // "No authentication required" + stream.write_all(&[0x05, 0x00]).await?; + + // Request: version, command, reserved, address type. + let mut request = [0u8; 4]; + stream.read_exact(&mut request).await?; + 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?; + let mut domain = vec![0u8; length[0] as usize]; + stream.read_exact(&mut domain).await?; + let mut port = [0u8; 2]; + stream.read_exact(&mut port).await?; + + let domain = + String::from_utf8(domain).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + let target = 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?; + + Ok((target, read_request(&mut stream).await?)) + } - let request = server.await.unwrap(); - 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)); + #[tokio::test] + 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")?; + assert!( + DefaultWebsocketTransport + .connect(&url, Some(proxy)) + .await + .is_err() + ); + + 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 19b6ad295..5a8d3d390 100644 --- a/nwc/CHANGELOG.md +++ b/nwc/CHANGELOG.md @@ -27,6 +27,12 @@ --> +## Unreleased + +### Breaking changes + +- Remove the `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` features + ## v0.45.1 - 2026/09/11 ### Fixed diff --git a/nwc/Cargo.toml b/nwc/Cargo.toml index 8553665c5..0fb1a0cb9 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/CHANGELOG.md b/signer/nostr-connect/CHANGELOG.md index a1dca24ca..6fed3e45d 100644 --- a/signer/nostr-connect/CHANGELOG.md +++ b/signer/nostr-connect/CHANGELOG.md @@ -27,6 +27,12 @@ --> +## Unreleased + +### Breaking changes + +- Remove the `native-tls`, `native-tls-vendored` and `rustls-tls-native-roots` features + ## v0.45.2 - 2026/09/11 ### Fixed diff --git a/signer/nostr-connect/Cargo.toml b/signer/nostr-connect/Cargo.toml index fb9ba8258..9c181f8bb 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]