From 3b54b30655dca4da4ffef70047ed4dce156e28b4 Mon Sep 17 00:00:00 2001 From: Noah Joeris Date: Wed, 26 Aug 2026 13:08:12 +0300 Subject: [PATCH 1/4] chore!: bump MSRV to 1.71 rustls 0.23 requires 1.71. This is crate-wide, so the following TLS feature is a breaking change for 1.70 toolchains. --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 5e3c99a..45088b2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ version = "0.4.0" description = "Experimental but sane electrum client by @evanlinjin." license = "MIT OR Apache-2.0" edition = "2021" -rust-version = "1.70" +rust-version = "1.71" repository = "https://github.com/bitcoindevkit/electrum_streaming_client" documentation = "https://docs.rs/electrum_streaming_client" readme = "README.md" From 10e85cc90e2e4b296a277541586037496383e485 Mon Sep 17 00:00:00 2001 From: Noah Joeris Date: Wed, 26 Aug 2026 13:09:29 +0300 Subject: [PATCH 2/4] feat(transport): add address types and plaintext TCP connect Keep hostnames unresolved for later TLS/SNI, and give blocking/tokio clients a connect-only TCP constructor. Dropping the blocking client shuts the TCP socket down so the read thread unblocks. --- Cargo.toml | 2 +- README.md | 13 +-- src/client.rs | 56 ++++++++++ src/lib.rs | 2 + src/transport/mod.rs | 246 +++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 312 insertions(+), 7 deletions(-) create mode 100644 src/transport/mod.rs diff --git a/Cargo.toml b/Cargo.toml index 45088b2..49bf4be 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,7 +15,7 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" bitcoin = { version = "0.32", features = ["serde"] } -tokio = { version = "1.44.2", features = ["io-util"], optional = true } +tokio = { version = "1.44.2", features = ["io-util", "net", "time"], optional = true } tokio-util = { version = "0.7.15", features = ["compat"], optional = true } [features] diff --git a/README.md b/README.md index 77bac7c..0399d98 100644 --- a/README.md +++ b/README.md @@ -16,15 +16,16 @@ models. ## Example (async with Tokio) ```rust,no_run -use electrum_streaming_client::{AsyncClient, Event}; -use tokio::net::TcpStream; +use std::time::Duration; + +use electrum_streaming_client::{AsyncClient, ServerAddr}; use futures::StreamExt; #[tokio::main] async fn main() -> anyhow::Result<()> { - let stream = TcpStream::connect("127.0.0.1:50001").await?; - let (reader, writer) = stream.into_split(); - let (client, mut events, worker) = AsyncClient::new_tokio(reader, writer); + let addr: ServerAddr = "127.0.0.1:50001".parse()?; + let (client, mut events, worker) = + AsyncClient::connect_tcp(&addr, Some(Duration::from_secs(10))).await?; tokio::spawn(worker); // spawn the client worker task @@ -41,7 +42,7 @@ async fn main() -> anyhow::Result<()> { ## Optional Features -- `tokio`: Enables [`AsyncClient::new_tokio`] for use with Tokio-compatible streams. +- `tokio`: Enables [`AsyncClient::new_tokio`] and [`AsyncClient::connect_tcp`]. ## License diff --git a/src/client.rs b/src/client.rs index 2c55200..dc9dc74 100644 --- a/src/client.rs +++ b/src/client.rs @@ -179,6 +179,21 @@ impl AsyncClient { self.tx.close_channel(); } + /// Creates a new [`AsyncClient`] connected to `addr` over plaintext TCP via Tokio. + #[cfg(feature = "tokio")] + pub async fn connect_tcp( + addr: &crate::transport::ServerAddr, + timeout: Option, + ) -> std::io::Result<( + Self, + AsyncEventReceiver, + impl std::future::Future> + Send, + )> { + let stream = crate::transport::tokio::connect_tcp(addr, timeout).await?; + let (reader, writer) = tokio::io::split(stream); + Ok(Self::new_tokio(reader, writer)) + } + /// Sends a single tracked request to the Electrum server and awaits the response. /// /// This method is for request–response style interactions where only a single result is @@ -256,6 +271,31 @@ impl AsyncClient { } } +/// A `Write` wrapper around a [`std::net::TcpStream`] that shuts down the socket on drop. +/// +/// [`BlockingClient::connect_tcp`] reads on a `try_clone` handle while the write thread holds +/// this wrapper. When the last client handle drops, the write thread ends and dropping this +/// wrapper shuts the socket down, unblocking the read thread. A plain `try_clone` handle would +/// otherwise keep the socket alive until the peer closes it. +#[derive(Debug)] +struct ShutdownOnDropTcpWriter(std::net::TcpStream); + +impl std::io::Write for ShutdownOnDropTcpWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + std::io::Write::write(&mut self.0, buf) + } + + fn flush(&mut self) -> std::io::Result<()> { + std::io::Write::flush(&mut self.0) + } +} + +impl Drop for ShutdownOnDropTcpWriter { + fn drop(&mut self) { + let _ = self.0.shutdown(std::net::Shutdown::Both); + } +} + /// A blocking Electrum client built on standard I/O. /// /// This client wraps a blocking transport implementing [`std::io::Read`] and [`std::io::Write`] and @@ -344,6 +384,22 @@ impl BlockingClient { (Self { tx: req_tx }, event_recv, read_join, write_join) } + /// Creates a new [`BlockingClient`] connected to `addr` over plaintext TCP. + #[allow(clippy::type_complexity)] + pub fn connect_tcp( + addr: &crate::transport::ServerAddr, + timeout: Option, + ) -> std::io::Result<( + Self, + BlockingEventReceiver, + std::thread::JoinHandle>, + std::thread::JoinHandle>, + )> { + let stream = crate::transport::blocking::connect_tcp(addr, timeout)?; + let reader = stream.try_clone()?; + Ok(Self::new(reader, ShutdownOnDropTcpWriter(stream))) + } + /// Sends a single tracked request to the Electrum server and waits for its response. /// /// This method blocks the current thread until the server replies. It is intended for diff --git a/src/lib.rs b/src/lib.rs index c48897e..f90e855 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -13,12 +13,14 @@ pub mod protocol; pub mod request; mod request_tracker; pub mod response; +pub mod transport; pub use hash_types::*; pub use pending_request::*; pub use protocol::*; pub use request::Request; pub use request_tracker::*; pub use serde_json; +pub use transport::{Host, ParseServerAddrError, ServerAddr}; /// An owned or borrowed static string. pub type CowStr = std::borrow::Cow<'static, str>; diff --git a/src/transport/mod.rs b/src/transport/mod.rs new file mode 100644 index 0000000..f7e5d7c --- /dev/null +++ b/src/transport/mod.rs @@ -0,0 +1,246 @@ +//! Address types and plaintext TCP constructors for Electrum connections. +//! +//! [`ServerAddr`] keeps the hostname unresolved so later TLS (SNI, certificates) can use the +//! original name. Use [`blocking::connect_tcp`] or [`tokio::connect_tcp`], or the client wrappers +//! [`crate::BlockingClient::connect_tcp`] / [`crate::AsyncClient::connect_tcp`]. + +use std::fmt; +use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; +use std::str::FromStr; + +/// The host portion of a [`ServerAddr`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Host { + /// A domain name, e.g. `electrum.example.com`. + Domain(String), + /// An IP literal. + Ip(IpAddr), +} + +impl fmt::Display for Host { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Host::Domain(domain) => f.write_str(domain), + Host::Ip(IpAddr::V4(ip)) => write!(f, "{}", ip), + Host::Ip(IpAddr::V6(ip)) => write!(f, "[{}]", ip), + } + } +} + +/// An Electrum server address: a [`Host`] and a port, without a connection scheme. +/// +/// Parses from `"host:port"`. IPv6 literals must be bracketed (`"[::1]:50001"`). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ServerAddr { + host: Host, + port: u16, +} + +impl ServerAddr { + /// Creates a new `ServerAddr` from a [`Host`] and port. + pub fn new(host: Host, port: u16) -> Self { + Self { host, port } + } + + /// The host portion of this address. + pub fn host(&self) -> &Host { + &self.host + } + + /// The port of this address. + pub fn port(&self) -> u16 { + self.port + } +} + +impl fmt::Display for ServerAddr { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}:{}", self.host, self.port) + } +} + +impl FromStr for ServerAddr { + type Err = ParseServerAddrError; + + fn from_str(s: &str) -> Result { + let invalid = || ParseServerAddrError(s.to_string()); + + if s.contains("://") { + return Err(invalid()); + } + + // Bracketed IP literal: "[]:". + if let Some(rest) = s.strip_prefix('[') { + let (ip_str, port_str) = rest.split_once("]:").ok_or_else(invalid)?; + return Ok(Self { + host: Host::Ip(ip_str.parse().map_err(|_| invalid())?), + port: port_str.parse().map_err(|_| invalid())?, + }); + } + + let (host_str, port_str) = s.rsplit_once(':').ok_or_else(invalid)?; + if host_str.is_empty() || host_str.contains(':') { + // Empty host, or an unbracketed IPv6 literal. + return Err(invalid()); + } + Ok(Self { + host: match host_str.parse::() { + Ok(ip) => Host::Ip(ip), + Err(_) => Host::Domain(host_str.to_string()), + }, + port: port_str.parse().map_err(|_| invalid())?, + }) + } +} + +impl ToSocketAddrs for ServerAddr { + type Iter = std::vec::IntoIter; + + /// Resolves this address via **local DNS**. + /// + /// Do not use this for `.onion` hosts — they are not resolvable via local DNS. + fn to_socket_addrs(&self) -> std::io::Result { + match &self.host { + Host::Ip(ip) => Ok(vec![SocketAddr::new(*ip, self.port)].into_iter()), + Host::Domain(domain) => (domain.as_str(), self.port).to_socket_addrs(), + } + } +} + +/// An error parsing a [`ServerAddr`] from a string. +/// +/// The payload is the full input string that failed to parse. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ParseServerAddrError(pub String); + +impl fmt::Display for ParseServerAddrError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "invalid server address '{}'", self.0) + } +} + +impl std::error::Error for ParseServerAddrError {} + +/// Blocking (std I/O) transport constructors. +pub mod blocking { + use std::io; + use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; + use std::time::Duration; + + use super::ServerAddr; + + /// Connects to `addr` over plaintext TCP using blocking I/O. + /// + /// `timeout` bounds TCP connection, not DNS. No read/write timeout + /// on the returned stream. + pub fn connect_tcp(addr: &ServerAddr, timeout: Option) -> io::Result { + let addrs: Vec<_> = addr.to_socket_addrs()?.collect(); + match timeout { + Some(timeout) => connect_with_total_timeout(&addrs, timeout), + None => TcpStream::connect(addrs.as_slice()), + } + } + + /// Tries each addr, splitting `timeout` across attempts. + fn connect_with_total_timeout( + addrs: &[SocketAddr], + mut timeout: Duration, + ) -> io::Result { + // Use the same algorithm as curl: 1/2 of the timeout on the first address, 1/4 on the + // second one, etc. https://curl.se/mail/lib-2014-11/0164.html + let mut last_err = None; + for (index, addr) in addrs.iter().enumerate() { + if index < addrs.len() - 1 { + timeout = timeout.div_f32(2.0); + } + match TcpStream::connect_timeout(addr, timeout) { + Ok(stream) => return Ok(stream), + Err(err) => last_err = Some(err), + } + } + Err(last_err.unwrap_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidInput, + "could not resolve to any addresses", + ) + })) + } +} + +/// Tokio-based async transport constructors. +#[cfg(feature = "tokio")] +pub mod tokio { + use std::io; + use std::net::SocketAddr; + use std::time::Duration; + + use tokio::net::TcpStream; + + use super::{Host, ServerAddr}; + + /// Connects to `addr` over plaintext TCP using the Tokio runtime. + /// + /// `timeout` bounds DNS and TCP connection. + pub async fn connect_tcp( + addr: &ServerAddr, + timeout: Option, + ) -> io::Result { + let connect_fut = async { + match addr.host() { + Host::Domain(domain) => TcpStream::connect((domain.as_str(), addr.port())).await, + Host::Ip(ip) => TcpStream::connect(SocketAddr::new(*ip, addr.port())).await, + } + }; + match timeout { + Some(timeout) => match tokio::time::timeout(timeout, connect_fut).await { + Ok(res) => res, + Err(_elapsed) => Err(io::Error::new( + io::ErrorKind::TimedOut, + format!("connection to '{}' timed out", addr), + )), + }, + None => connect_fut.await, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; + + fn addr(s: &str) -> ServerAddr { + s.parse().unwrap_or_else(|e| panic!("{s:?}: {e}")) + } + + #[test] + fn server_addr_parse() { + let a = addr("127.0.0.1:50001"); + assert_eq!(a.host(), &Host::Ip(IpAddr::V4(Ipv4Addr::LOCALHOST))); + assert_eq!(a.port(), 50001); + assert_eq!(a.to_string(), "127.0.0.1:50001"); + + let a = addr("localhost:50001"); + assert_eq!(a.host(), &Host::Domain("localhost".into())); + assert_eq!(a.port(), 50001); + + let a = addr("[::1]:50001"); + assert_eq!(a.host(), &Host::Ip(IpAddr::V6(Ipv6Addr::LOCALHOST))); + assert_eq!(a.port(), 50001); + assert_eq!(a.to_string(), "[::1]:50001"); + + for bad in [ + "tcp://127.0.0.1:50001", + "ssl://host:50002", + "::1:50001", + ":50001", + "host", + "host:", + "host:99999", + "[::1]50001", + "[example.com]:50001", + ] { + assert!(bad.parse::().is_err(), "{bad}"); + } + } +} From fecfba54ae2bd3cc431ca91ef45d8eef6600352b Mon Sep 17 00:00:00 2001 From: Noah Joeris Date: Wed, 26 Aug 2026 13:20:34 +0300 Subject: [PATCH 3/4] feat(transport): add TLS Optional `ssl` feature using rustls 0.23. Blocking and tokio clients get `connect_ssl`. Flush JSON-RPC writes so TLS buffers are pushed. --- Cargo.lock | 327 +++++++++++++++++++++++------------- Cargo.toml | 7 + README.md | 1 + src/client.rs | 46 ++++++ src/io.rs | 9 +- src/lib.rs | 5 +- src/transport/mod.rs | 385 ++++++++++++++++++++++++++++++++++++++++++- src/transport/tls.rs | 308 ++++++++++++++++++++++++++++++++++ 8 files changed, 970 insertions(+), 118 deletions(-) create mode 100644 src/transport/tls.rs diff --git a/Cargo.lock b/Cargo.lock index 24b483e..749f5de 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -53,9 +53,9 @@ dependencies = [ [[package]] name = "async-channel" -version = "2.3.1" +version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89b47800b0be77592da0afd425cc03468052844aff33b84e33cc696f64e77b6a" +checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2" dependencies = [ "concurrent-queue", "event-listener-strategy", @@ -65,9 +65,9 @@ dependencies = [ [[package]] name = "async-executor" -version = "1.13.2" +version = "1.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb812ffb58524bdd10860d7d974e2f01cc0950c2438a74ee5ec2e2280c6c4ffa" +checksum = "c96bf972d85afc50bf5ab8fe2d54d1586b4e0b46c97c50a0c9e71e2f7bcd812a" dependencies = [ "async-task", "concurrent-queue", @@ -83,7 +83,7 @@ version = "2.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05b1b633a2115cd122d73b955eadd9916c18c8f510ec9cd1686404c60ad1c29c" dependencies = [ - "async-channel 2.3.1", + "async-channel 2.5.0", "async-executor", "async-io", "async-lock", @@ -94,11 +94,11 @@ dependencies = [ [[package]] name = "async-io" -version = "2.4.1" +version = "2.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1237c0ae75a0f3765f58910ff9cdd0a12eeb39ab2f4c7de23262f337f0aacbb3" +checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc" dependencies = [ - "async-lock", + "autocfg", "cfg-if", "concurrent-queue", "futures-io", @@ -107,26 +107,25 @@ dependencies = [ "polling", "rustix 1.0.7", "slab", - "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] name = "async-lock" -version = "3.4.0" +version = "3.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff6e472cdea888a4bd64f342f09b3f50e1886d32afe8df3d663c01140b811b18" +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" dependencies = [ - "event-listener 5.4.0", + "event-listener 5.4.2", "event-listener-strategy", "pin-project-lite", ] [[package]] name = "async-std" -version = "1.13.1" +version = "1.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "730294c1c08c2e0f85759590518f6333f0d5a0a766a27d519c1b244c3dfd8a24" +checksum = "2c8e079a4ab67ae52b7403632e4618815d6db36d2a010cfe41b02c1b1578f93b" dependencies = [ "async-channel 1.9.0", "async-global-executor", @@ -166,6 +165,29 @@ version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26" +[[package]] +name = "aws-lc-rs" +version = "1.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.44.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f09fae7be8bb3174e05c6afdb34199e6dc0c7c04ba9fa237b1967adfbde27483" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", + "pkg-config", +] + [[package]] name = "backtrace" version = "0.3.75" @@ -358,11 +380,11 @@ dependencies = [ [[package]] name = "blocking" -version = "1.6.1" +version = "1.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "703f41c54fc768e63e091340b424302bb1c29ef4aa0c7f10fe849dfb114d29ea" +checksum = "e83f8d02be6967315521be875afa792a316e28d57b5a2d401897e2a7921b7f21" dependencies = [ - "async-channel 2.3.1", + "async-channel 2.5.0", "async-task", "futures-io", "futures-lite", @@ -371,9 +393,9 @@ dependencies = [ [[package]] name = "bumpalo" -version = "3.17.0" +version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1628fb46dfa0b37568d12e5edd512553eccf6a22a78e8bde00bb4aed84d5bdbf" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" [[package]] name = "byteorder" @@ -409,10 +431,11 @@ dependencies = [ [[package]] name = "cc" -version = "1.2.25" +version = "1.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d0fc897dc1e865cc67c0e05a836d9d3f1df3cbe442aa4a9473b18e12624a4951" +checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d" dependencies = [ + "find-msvc-tools", "jobserver", "libc", "shlex", @@ -434,6 +457,15 @@ dependencies = [ "inout", ] +[[package]] +name = "cmake" +version = "0.1.58" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" +dependencies = [ + "cc", +] + [[package]] name = "concurrent-queue" version = "2.5.0" @@ -503,6 +535,12 @@ dependencies = [ "subtle", ] +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "either" version = "1.15.0" @@ -546,10 +584,13 @@ dependencies = [ "bdk_testenv", "bitcoin", "futures", + "rustls 0.23.43", "serde", "serde_json", "tokio", + "tokio-rustls", "tokio-util", + "webpki-roots 1.0.9", ] [[package]] @@ -570,11 +611,10 @@ checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" [[package]] name = "event-listener" -version = "5.4.0" +version = "5.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3492acde4c3fc54c845eaab3eed8bd00c7a7d881f78bfc801e43a93dec1331ae" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" dependencies = [ - "concurrent-queue", "parking", "pin-project-lite", ] @@ -585,7 +625,7 @@ version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" dependencies = [ - "event-listener 5.4.0", + "event-listener 5.4.2", "pin-project-lite", ] @@ -607,6 +647,12 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "find-msvc-tools" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" + [[package]] name = "flate2" version = "1.1.1" @@ -617,6 +663,12 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "futures" version = "0.3.31" @@ -668,9 +720,9 @@ checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" [[package]] name = "futures-lite" -version = "2.6.0" +version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f5edaec856126859abb19ed65f39e90fea3a9574b9707f13539acf4abf7eb532" +checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" dependencies = [ "fastrand", "futures-core", @@ -687,7 +739,7 @@ checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.101", ] [[package]] @@ -773,9 +825,9 @@ dependencies = [ [[package]] name = "hermit-abi" -version = "0.5.1" +version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f154ce46856750ed433c8649605bf7ed2de3bc35fd9d2a9f30cddd873c80cb08" +checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" [[package]] name = "hex-conservative" @@ -837,11 +889,12 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.77" +version = "0.3.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1cfaf33c695fc6e08064efbc1f72ec937429614f25eef83af942d0e227c3a28f" +checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" dependencies = [ - "once_cell", + "cfg-if", + "futures-util", "wasm-bindgen", ] @@ -955,11 +1008,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0d2aaba477837b46ec1289588180fabfccf0c3b1d1a0c6b1866240cd6cd5ce9" dependencies = [ "log", - "rustls", - "rustls-webpki", + "rustls 0.21.12", + "rustls-webpki 0.101.7", "serde", "serde_json", - "webpki-roots", + "webpki-roots 0.25.4", ] [[package]] @@ -1084,9 +1137,9 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" [[package]] name = "piper" -version = "0.2.4" +version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96c8c490f422ef9a4efd2cb5b42b76c8613d7e7dfc1caf667b8a3350a5acc066" +checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" dependencies = [ "atomic-waker", "fastrand", @@ -1101,17 +1154,16 @@ checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" [[package]] name = "polling" -version = "3.8.0" +version = "3.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b53a684391ad002dd6a596ceb6c74fd004fdce75f4be2e3f615068abbea5fd50" +checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" dependencies = [ "cfg-if", "concurrent-queue", "hermit-abi", "pin-project-lite", "rustix 1.0.7", - "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -1246,10 +1298,34 @@ checksum = "3f56a14d1f48b391359b22f731fd4bd7e43c97f3c50eee276f3aa09c94784d3e" dependencies = [ "log", "ring", - "rustls-webpki", + "rustls-webpki 0.101.7", "sct", ] +[[package]] +name = "rustls" +version = "0.23.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +dependencies = [ + "aws-lc-rs", + "log", + "once_cell", + "rustls-pki-types", + "rustls-webpki 0.103.14", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + [[package]] name = "rustls-webpki" version = "0.101.7" @@ -1260,11 +1336,23 @@ dependencies = [ "untrusted", ] +[[package]] +name = "rustls-webpki" +version = "0.103.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0527518605e68109d875e248ea259b6758801cf165e4b2c2733ae3b51f12535a" +dependencies = [ + "aws-lc-rs", + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" -version = "1.0.21" +version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8a0d197bd2c9dc6e53b84da9556a69ba4cdfab8619eb41a8bd1cc2027a0f6b1d" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" [[package]] name = "ryu" @@ -1311,22 +1399,32 @@ dependencies = [ [[package]] name = "serde" -version = "1.0.219" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f0e2c6ed6606019b4e29e69dbaba95b11854410e5347d525002456dbbb786b6" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.219" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b0276cf7f2c73365f7157c8123c21cd9a50fbbd844757af28ca1f5925fc2a00" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.3", ] [[package]] @@ -1365,9 +1463,9 @@ dependencies = [ [[package]] name = "shlex" -version = "1.3.0" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" [[package]] name = "signal-hook-registry" @@ -1420,6 +1518,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "tar" version = "0.4.44" @@ -1489,7 +1598,17 @@ checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.101", +] + +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls 0.23.43", + "tokio", ] [[package]] @@ -1506,22 +1625,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "tracing" -version = "0.1.41" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0" -dependencies = [ - "pin-project-lite", - "tracing-core", -] - -[[package]] -name = "tracing-core" -version = "0.1.33" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e672c95779cf947c5311f83787af4fa8fffd12fb27e4993211a84bdfd9610f9c" - [[package]] name = "typenum" version = "1.18.0" @@ -1542,9 +1645,9 @@ checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" [[package]] name = "value-bag" -version = "1.11.1" +version = "1.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "943ce29a8a743eb10d6082545d861b24f9d1b160b7d741e0f2cdf726bec909c5" +checksum = "068e763e8279de7ab94b6afebded2cb701678af094feb1c12ccb061b4783c1be" [[package]] name = "version_check" @@ -1569,48 +1672,32 @@ dependencies = [ [[package]] name = "wasm-bindgen" -version = "0.2.100" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1edc8929d7499fc4e8f0be2262a241556cfc54a0bea223790e71446f2aab1ef5" +checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" dependencies = [ "cfg-if", "once_cell", "rustversion", "wasm-bindgen-macro", -] - -[[package]] -name = "wasm-bindgen-backend" -version = "0.2.100" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f0a0651a5c2bc21487bde11ee802ccaf4c51935d0d3d42a6101f98161700bc6" -dependencies = [ - "bumpalo", - "log", - "proc-macro2", - "quote", - "syn", "wasm-bindgen-shared", ] [[package]] name = "wasm-bindgen-futures" -version = "0.4.50" +version = "0.4.76" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "555d470ec0bc3bb57890405e5d4322cc9ea83cebb085523ced7be4144dac1e61" +checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d" dependencies = [ - "cfg-if", "js-sys", - "once_cell", "wasm-bindgen", - "web-sys", ] [[package]] name = "wasm-bindgen-macro" -version = "0.2.100" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7fe63fc6d09ed3792bd0897b314f53de8e16568c2b3f7982f468c0bf9bd0b407" +checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -1618,41 +1705,40 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.100" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de" +checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" dependencies = [ + "bumpalo", "proc-macro2", "quote", - "syn", - "wasm-bindgen-backend", + "syn 2.0.101", "wasm-bindgen-shared", ] [[package]] name = "wasm-bindgen-shared" -version = "0.2.100" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a05d73b933a847d6cccdda8f838a22ff101ad9bf93e33684f39c1f5f0eece3d" +checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" dependencies = [ "unicode-ident", ] [[package]] -name = "web-sys" -version = "0.3.77" +name = "webpki-roots" +version = "0.25.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33b6dd2ef9186f1f2072e409e99cd22a975331a6b3591b12c764e0e55c60d5d2" -dependencies = [ - "js-sys", - "wasm-bindgen", -] +checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" [[package]] name = "webpki-roots" -version = "0.25.4" +version = "1.0.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" +checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +dependencies = [ + "rustls-pki-types", +] [[package]] name = "which" @@ -1666,6 +1752,12 @@ dependencies = [ "rustix 0.38.44", ] +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + [[package]] name = "windows-sys" version = "0.52.0" @@ -1684,6 +1776,15 @@ dependencies = [ "windows-targets", ] +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-targets" version = "0.52.6" @@ -1784,9 +1885,15 @@ checksum = "28a6e20d751156648aa063f3800b706ee209a32c0b4d9f24be3d980b01be55ef" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.101", ] +[[package]] +name = "zeroize" +version = "1.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" + [[package]] name = "zip" version = "0.6.6" diff --git a/Cargo.toml b/Cargo.toml index 49bf4be..74a9186 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,9 @@ repository = "https://github.com/bitcoindevkit/electrum_streaming_client" documentation = "https://docs.rs/electrum_streaming_client" readme = "README.md" +[package.metadata.docs.rs] +all-features = true + [dependencies] futures = "0.3" serde = { version = "1", features = ["derive"] } @@ -17,10 +20,14 @@ bitcoin = { version = "0.32", features = ["serde"] } tokio = { version = "1.44.2", features = ["io-util", "net", "time"], optional = true } tokio-util = { version = "0.7.15", features = ["compat"], optional = true } +tokio-rustls = { version = "0.26", optional = true } +rustls = { version = "0.23", optional = true } +webpki-roots = { version = "1", optional = true } [features] default = ["tokio"] tokio = ["dep:tokio", "tokio-util"] +ssl = ["dep:rustls", "dep:webpki-roots", "dep:tokio-rustls"] [dev-dependencies] async-std = "1.13.0" diff --git a/README.md b/README.md index 0399d98..c4742ae 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,7 @@ async fn main() -> anyhow::Result<()> { ## Optional Features - `tokio`: Enables [`AsyncClient::new_tokio`] and [`AsyncClient::connect_tcp`]. +- `ssl`: Enables TLS via rustls. Async TLS additionally requires `tokio`. ## License diff --git a/src/client.rs b/src/client.rs index dc9dc74..63ed26f 100644 --- a/src/client.rs +++ b/src/client.rs @@ -194,6 +194,28 @@ impl AsyncClient { Ok(Self::new_tokio(reader, writer)) } + /// Creates a new [`AsyncClient`] connected to `addr` over TLS via Tokio. + /// + /// `timeout` bounds DNS, TCP connect, and the TLS handshake. + /// `validate_domain` requires a domain host. + #[cfg(all(feature = "ssl", feature = "tokio"))] + pub async fn connect_ssl( + addr: &crate::transport::ServerAddr, + validate_domain: bool, + timeout: Option, + ) -> Result< + ( + Self, + AsyncEventReceiver, + impl std::future::Future> + Send, + ), + crate::ConnectError, + > { + let stream = crate::transport::tokio::connect_ssl(addr, validate_domain, timeout).await?; + let (reader, writer) = tokio::io::split(stream); + Ok(Self::new_tokio(reader, writer)) + } + /// Sends a single tracked request to the Electrum server and awaits the response. /// /// This method is for request–response style interactions where only a single result is @@ -400,6 +422,30 @@ impl BlockingClient { Ok(Self::new(reader, ShutdownOnDropTcpWriter(stream))) } + /// Creates a new [`BlockingClient`] connected to `addr` over TLS. + /// + /// `timeout` bounds TCP connect and the TLS handshake. + /// `validate_domain` requires a domain host. + #[cfg(feature = "ssl")] + #[allow(clippy::type_complexity)] + pub fn connect_ssl( + addr: &crate::transport::ServerAddr, + validate_domain: bool, + timeout: Option, + ) -> Result< + ( + Self, + BlockingEventReceiver, + std::thread::JoinHandle>, + std::thread::JoinHandle>, + ), + crate::ConnectError, + > { + let stream = crate::transport::blocking::connect_ssl(addr, validate_domain, timeout)?; + let (reader, writer) = stream.into_split(); + Ok(Self::new(reader, writer)) + } + /// Sends a single tracked request to the Electrum server and waits for its response. /// /// This method blocks the current thread until the server replies. It is intended for diff --git a/src/io.rs b/src/io.rs index 58a83ec..3518870 100644 --- a/src/io.rs +++ b/src/io.rs @@ -178,7 +178,8 @@ where { let mut b = serde_json::to_vec(&msg.into()).expect("must serialize"); b.push(b'\n'); - writer.write_all(&b) + writer.write_all(&b)?; + writer.flush() } /// Asynchronously writes a JSON-RPC request or batch to an async writer, followed by a newline. @@ -200,7 +201,8 @@ where use futures::AsyncWriteExt; let mut b = serde_json::to_vec(&msg.into()).expect("must serialize"); b.push(b'\n'); - writer.write_all(&b).await + writer.write_all(&b).await?; + writer.flush().await } /// Asynchronously writes a JSON-RPC request or batch to a tokio async writer, followed by a newline. @@ -215,5 +217,6 @@ where use tokio::io::AsyncWriteExt; let mut b = serde_json::to_vec(&msg.into()).expect("must serialize"); b.push(b'\n'); - writer.write_all(&b).await + writer.write_all(&b).await?; + writer.flush().await } diff --git a/src/lib.rs b/src/lib.rs index f90e855..4ab719d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,7 +20,10 @@ pub use protocol::*; pub use request::Request; pub use request_tracker::*; pub use serde_json; -pub use transport::{Host, ParseServerAddrError, ServerAddr}; +pub use transport::{ConnectError, Host, ParseServerAddrError, ServerAddr}; + +#[cfg(feature = "ssl")] +pub use transport::TlsError; /// An owned or borrowed static string. pub type CowStr = std::borrow::Cow<'static, str>; diff --git a/src/transport/mod.rs b/src/transport/mod.rs index f7e5d7c..0a66010 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -1,12 +1,18 @@ -//! Address types and plaintext TCP constructors for Electrum connections. +//! Address types and TCP/TLS constructors for Electrum connections. //! -//! [`ServerAddr`] keeps the hostname unresolved so later TLS (SNI, certificates) can use the -//! original name. Use [`blocking::connect_tcp`] or [`tokio::connect_tcp`], or the client wrappers -//! [`crate::BlockingClient::connect_tcp`] / [`crate::AsyncClient::connect_tcp`]. +//! [`ServerAddr`] keeps the hostname unresolved so TLS SNI and certificate validation can use +//! the original name. Use [`blocking::connect_tcp`] / [`blocking::connect_ssl`] or the tokio +//! equivalents, or the client wrappers [`crate::BlockingClient::connect_tcp`] / +//! [`crate::AsyncClient::connect_tcp`]. use std::fmt; use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; use std::str::FromStr; +#[cfg(feature = "ssl")] +use std::sync::Arc; + +#[cfg(feature = "ssl")] +mod tls; /// The host portion of a [`ServerAddr`]. #[derive(Debug, Clone, PartialEq, Eq)] @@ -121,11 +127,242 @@ impl fmt::Display for ParseServerAddrError { impl std::error::Error for ParseServerAddrError {} +/// Error establishing an Electrum connection. +#[non_exhaustive] +#[derive(Debug)] +pub enum ConnectError { + /// Transport or socket failure, including DNS, TCP, timeout, and handshake EOF/reset. + Io(std::io::Error), + /// TLS configuration, protocol, or certificate-validation failure. + #[cfg(feature = "ssl")] + Tls(TlsError), +} + +#[cfg(feature = "ssl")] +impl ConnectError { + /// Returns the certificate-validation error, if present. + pub fn certificate_error(&self) -> Option<&rustls::CertificateError> { + match self { + ConnectError::Tls(e) => e.certificate_error(), + _ => None, + } + } + + /// Classifies embedded rustls errors as TLS failures and preserves other I/O errors. + fn from_handshake_io(err: std::io::Error) -> Self { + match err + .get_ref() + .and_then(|inner| inner.downcast_ref::()) + { + Some(tls) => ConnectError::Tls(TlsError::Rustls(tls.clone())), + None => ConnectError::Io(err), + } + } +} + +impl fmt::Display for ConnectError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ConnectError::Io(e) => write!(f, "connection I/O error: {e}"), + #[cfg(feature = "ssl")] + ConnectError::Tls(e) => write!(f, "{e}"), + } + } +} + +impl std::error::Error for ConnectError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + ConnectError::Io(e) => Some(e), + #[cfg(feature = "ssl")] + ConnectError::Tls(e) => Some(e), + } + } +} + +impl From for ConnectError { + fn from(e: std::io::Error) -> Self { + ConnectError::Io(e) + } +} + +#[cfg(feature = "ssl")] +impl From for ConnectError { + fn from(e: TlsError) -> Self { + ConnectError::Tls(e) + } +} + +/// Error in the TLS layer of a connection: configuration, protocol, or +/// certificate-validation failures. Transport failures are [`ConnectError::Io`]. +#[cfg(feature = "ssl")] +#[non_exhaustive] +#[derive(Debug)] +pub enum TlsError { + /// TLS protocol or certificate-validation failure. + Rustls(rustls::Error), + /// `validate_domain` is true and the host is [`Host::Ip`]. + MissingDomain, + /// Host string is not a valid TLS [`rustls::pki_types::ServerName`]. + InvalidServerName(String), +} + +#[cfg(feature = "ssl")] +impl TlsError { + /// Server certificate rejected by the local verifier. + pub fn certificate_error(&self) -> Option<&rustls::CertificateError> { + match self { + Self::Rustls(rustls::Error::InvalidCertificate(error)) => Some(error), + _ => None, + } + } +} + +#[cfg(feature = "ssl")] +impl fmt::Display for TlsError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + TlsError::MissingDomain => { + write!(f, "TLS certificate validation requires a domain name") + } + TlsError::InvalidServerName(name) => { + write!(f, "invalid TLS server name '{name}'") + } + TlsError::Rustls(e) => write!(f, "TLS error: {e}"), + } + } +} + +#[cfg(feature = "ssl")] +impl std::error::Error for TlsError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + TlsError::Rustls(e) => Some(e), + TlsError::MissingDomain | TlsError::InvalidServerName(_) => None, + } + } +} + +#[cfg(feature = "ssl")] +impl From for TlsError { + fn from(e: rustls::Error) -> Self { + TlsError::Rustls(e) + } +} + +#[cfg(feature = "ssl")] +impl From for ConnectError { + fn from(e: rustls::Error) -> Self { + ConnectError::Tls(TlsError::Rustls(e)) + } +} + +/// Process-default provider if the application installed one; otherwise aws-lc-rs. +#[cfg(feature = "ssl")] +fn crypto_provider() -> Arc { + rustls::crypto::CryptoProvider::get_default() + .cloned() + .unwrap_or_else(|| Arc::new(rustls::crypto::aws_lc_rs::default_provider())) +} + +#[cfg(feature = "ssl")] +fn client_config(validate_domain: bool) -> Result, rustls::Error> { + let provider = crypto_provider(); + let builder = rustls::ClientConfig::builder_with_provider(Arc::clone(&provider)) + .with_safe_default_protocol_versions()?; + let config = if validate_domain { + let roots = rustls::RootCertStore { + roots: webpki_roots::TLS_SERVER_ROOTS.to_vec(), + }; + builder.with_root_certificates(roots).with_no_client_auth() + } else { + builder + .dangerous() + .with_custom_certificate_verifier(Arc::new(danger::NoCertificateVerification::new( + (*provider).clone(), + ))) + .with_no_client_auth() + }; + Ok(Arc::new(config)) +} + +#[cfg(feature = "ssl")] +fn server_name( + addr: &ServerAddr, + validate_domain: bool, +) -> Result, TlsError> { + match addr.host() { + Host::Domain(domain) => rustls::pki_types::ServerName::try_from(domain.clone()) + .map_err(|_| TlsError::InvalidServerName(domain.clone())), + Host::Ip(ip) => { + if validate_domain { + Err(TlsError::MissingDomain) + } else { + Ok(rustls::pki_types::ServerName::from(*ip)) + } + } + } +} + +#[cfg(feature = "ssl")] +mod danger { + use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified}; + use rustls::crypto::CryptoProvider; + use rustls::pki_types::{CertificateDer, ServerName, UnixTime}; + use rustls::DigitallySignedStruct; + + #[derive(Debug)] + pub struct NoCertificateVerification(CryptoProvider); + + impl NoCertificateVerification { + pub fn new(provider: CryptoProvider) -> Self { + Self(provider) + } + } + + impl rustls::client::danger::ServerCertVerifier for NoCertificateVerification { + fn verify_server_cert( + &self, + _end_entity: &CertificateDer<'_>, + _intermediates: &[CertificateDer<'_>], + _server_name: &ServerName<'_>, + _ocsp: &[u8], + _now: UnixTime, + ) -> Result { + Ok(ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + _message: &[u8], + _cert: &CertificateDer<'_>, + _dss: &DigitallySignedStruct, + ) -> Result { + Ok(HandshakeSignatureValid::assertion()) + } + + fn verify_tls13_signature( + &self, + _message: &[u8], + _cert: &CertificateDer<'_>, + _dss: &DigitallySignedStruct, + ) -> Result { + Ok(HandshakeSignatureValid::assertion()) + } + + fn supported_verify_schemes(&self) -> Vec { + self.0.signature_verification_algorithms.supported_schemes() + } + } +} + /// Blocking (std I/O) transport constructors. pub mod blocking { use std::io; use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; use std::time::Duration; + #[cfg(feature = "ssl")] + use std::time::Instant; use super::ServerAddr; @@ -141,6 +378,113 @@ pub mod blocking { } } + #[cfg(feature = "ssl")] + pub use super::tls::{TlsReadHalf, TlsStream, TlsWriteHalf}; + + /// Connects to `addr` over TLS using blocking I/O. + /// + /// `timeout` is a single deadline covering TCP connect and the TLS handshake. + /// The stream can be used as a single `Read + Write`, or split with + /// [`TlsStream::into_split`] so a reader and writer can run on separate threads. + #[cfg(feature = "ssl")] + pub fn connect_ssl( + addr: &ServerAddr, + validate_domain: bool, + timeout: Option, + ) -> Result { + let deadline = timeout.map(|timeout| Instant::now() + timeout); + let server_name = super::server_name(addr, validate_domain)?; + let mut tcp = connect_tcp(addr, timeout).map_err(super::ConnectError::Io)?; + + let mut conn = + rustls::ClientConnection::new(super::client_config(validate_domain)?, server_name)?; + conn.complete_io(&mut HandshakeIo { + tcp: &mut tcp, + deadline, + }) + .map_err(super::ConnectError::from_handshake_io)?; + + tcp.set_read_timeout(None) + .map_err(super::ConnectError::Io)?; + tcp.set_write_timeout(None) + .map_err(super::ConnectError::Io)?; + TlsStream::new(conn, tcp).map_err(super::ConnectError::Io) + } + + /// Remaining time until `deadline`, or `TimedOut` if it has passed. + #[cfg(feature = "ssl")] + fn remaining_timeout(deadline: Instant) -> io::Result { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + Err(io::Error::new( + io::ErrorKind::TimedOut, + "TLS handshake timed out", + )) + } else { + Ok(remaining) + } + } + + /// `Read + Write` adapter for rustls `complete_io`, enforcing a single + /// deadline for the whole handshake. + /// + /// Without it, socket timeouts would apply per operation, so the handshake + /// could take several times longer than the intended deadline. + #[cfg(feature = "ssl")] + struct HandshakeIo<'a> { + tcp: &'a mut TcpStream, + deadline: Option, + } + + /// Make socket timeouts readable as `TimedOut` rather than rustls's + /// bare `WouldBlock` propagation. + #[cfg(feature = "ssl")] + fn map_timeout(deadline: Option, result: io::Result) -> io::Result { + match result { + Err(e) if e.kind() == io::ErrorKind::WouldBlock && deadline.is_some() => { + Err(io::Error::new(io::ErrorKind::TimedOut, e)) + } + other => other, + } + } + + #[cfg(feature = "ssl")] + impl HandshakeIo<'_> { + fn apply_deadline(&mut self) -> io::Result<()> { + let Some(deadline) = self.deadline else { + return Ok(()); + }; + let remaining = remaining_timeout(deadline)?; + self.tcp.set_read_timeout(Some(remaining))?; + self.tcp.set_write_timeout(Some(remaining))?; + Ok(()) + } + } + + #[cfg(feature = "ssl")] + impl io::Read for HandshakeIo<'_> { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.apply_deadline()?; + let result = self.tcp.read(buf); + map_timeout(self.deadline, result) + } + } + + #[cfg(feature = "ssl")] + impl io::Write for HandshakeIo<'_> { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.apply_deadline()?; + let result = self.tcp.write(buf); + map_timeout(self.deadline, result) + } + + fn flush(&mut self) -> io::Result<()> { + self.apply_deadline()?; + let result = self.tcp.flush(); + map_timeout(self.deadline, result) + } + } + /// Tries each addr, splitting `timeout` across attempts. fn connect_with_total_timeout( addrs: &[SocketAddr], @@ -202,6 +546,39 @@ pub mod tokio { None => connect_fut.await, } } + + /// Connects to `addr` over TLS using the Tokio runtime. + /// + /// `timeout` bounds DNS, TCP connect, and the TLS handshake. + /// `validate_domain` requires a [`super::Host::Domain`]. + #[cfg(feature = "ssl")] + pub async fn connect_ssl( + addr: &ServerAddr, + validate_domain: bool, + timeout: Option, + ) -> Result, super::ConnectError> { + let server_name = super::server_name(addr, validate_domain)?; + let connector = tokio_rustls::TlsConnector::from(super::client_config(validate_domain)?); + let handshake = async { + let tcp = connect_tcp(addr, None) + .await + .map_err(super::ConnectError::Io)?; + connector + .connect(server_name, tcp) + .await + .map_err(super::ConnectError::from_handshake_io) + }; + match timeout { + Some(timeout) => match tokio::time::timeout(timeout, handshake).await { + Ok(res) => res, + Err(_elapsed) => Err(super::ConnectError::Io(io::Error::new( + io::ErrorKind::TimedOut, + format!("connection to '{}' timed out", addr), + ))), + }, + None => handshake.await, + } + } } #[cfg(test)] diff --git a/src/transport/tls.rs b/src/transport/tls.rs new file mode 100644 index 0000000..4a5a422 --- /dev/null +++ b/src/transport/tls.rs @@ -0,0 +1,308 @@ +//! Split-capable blocking TLS stream built on rustls. +//! +//! `rustls::StreamOwned` cannot provide independent read and write halves. This adapter shares +//! the rustls connection state but releases its lock before blocking on socket I/O, allowing +//! reads and writes to progress independently. +//! +//! When both locks are needed, acquire TLS output before the rustls connection state. + +use std::io::{self, Read, Write}; +use std::net::{Shutdown, TcpStream}; +use std::sync::{Arc, Mutex, MutexGuard}; + +const CIPHERTEXT_CHUNK: usize = 16 * 1024; + +#[derive(Debug)] +struct TlsOutput { + /// Destination for queued TLS ciphertext. + writer: W, + /// TLS ciphertext from rustls before sending it to `writer`. + /// `offset` marks the already-written prefix. + pending_ciphertext: Vec, + /// Number of leading bytes in `pending_ciphertext` already accepted by `writer`. + offset: usize, +} + +impl TlsOutput { + fn new(writer: W) -> Self { + Self { + writer, + pending_ciphertext: Vec::new(), + offset: 0, + } + } + + /// Queues new ciphertext and writes all pending ciphertext. + fn queue_and_write_ciphertext(&mut self, ciphertext: &[u8]) -> io::Result<()> { + self.pending_ciphertext.extend_from_slice(ciphertext); + while self.offset < self.pending_ciphertext.len() { + match self.writer.write(&self.pending_ciphertext[self.offset..]) { + Ok(0) => return Err(io::Error::from(io::ErrorKind::WriteZero)), + Ok(n) => self.offset += n, + Err(err) if err.kind() == io::ErrorKind::Interrupted => continue, + Err(err) => return Err(err), + } + } + self.pending_ciphertext.clear(); + self.offset = 0; + Ok(()) + } + + fn flush(&mut self) -> io::Result<()> { + self.queue_and_write_ciphertext(&[])?; + self.writer.flush() + } +} + +#[derive(Debug)] +struct Shared { + /// TLS protocol state shared by the read and write halves. + tls_connection: Mutex, + /// Retains and sends TLS ciphertext produced by `tls_connection`. + /// The read half also writes here via `emit` for TLS alerts and KeyUpdate responses. + tls_output: Mutex>, + /// Extra handle used by either half's `Drop` to interrupt blocking socket I/O. + /// It avoids waiting on `tls_output`, which a blocked writer may hold locked. + shutdown_socket: TcpStream, +} + +impl Shared { + fn tls_connection(&self) -> io::Result> { + self.tls_connection.lock().map_err(|_poison| { + io::Error::new(io::ErrorKind::Other, "TLS connection state mutex poisoned") + }) + } + + fn tls_output(&self) -> io::Result>> { + self.tls_output + .lock() + .map_err(|_poison| io::Error::new(io::ErrorKind::Other, "TLS output mutex poisoned")) + } + + /// Sends queued TLS records, releasing TLS state first so reads can continue. + fn emit(&self) -> io::Result<()> { + let mut output = self.tls_output()?; + let ciphertext = { + let mut conn = self.tls_connection()?; + let mut ciphertext = Vec::new(); + drain_tls(&mut conn, &mut ciphertext)?; + ciphertext + }; + output.queue_and_write_ciphertext(&ciphertext)?; + output.flush() + } +} + +/// Blocking TLS stream over a TCP socket. +/// +/// Use [`into_split`](Self::into_split) to give the read and write threads independent halves. +/// A successful write may still have ciphertext queued; use [`Write::flush`] to report +/// pending socket errors. +#[derive(Debug)] +pub struct TlsStream { + reader: TlsReadHalf, + writer: TlsWriteHalf, +} + +/// Read half of a split [`TlsStream`]. +/// +/// Dropping this half shuts down the TCP socket so the writer unblocks. +#[derive(Debug)] +pub struct TlsReadHalf { + shared: Arc, + read_socket: TcpStream, + /// Inbound TCP bytes not yet consumed by `read_tls`. Incomplete records live in rustls. + pending_incoming: Vec, +} + +/// Write half of a split [`TlsStream`]. +/// +/// Dropping this half shuts down the TCP socket so the reader unblocks. +/// A successful write may still have ciphertext queued; use [`Write::flush`] to report +/// pending socket errors. +#[derive(Debug)] +pub struct TlsWriteHalf { + shared: Arc, +} + +impl TlsStream { + /// `conn` must already have completed the handshake. + pub(super) fn new(conn: rustls::ClientConnection, tcp: TcpStream) -> io::Result { + let read_socket = tcp.try_clone()?; + let write_socket = tcp.try_clone()?; + let shared = Arc::new(Shared { + tls_connection: Mutex::new(conn), + tls_output: Mutex::new(TlsOutput::new(write_socket)), + shutdown_socket: tcp, + }); + Ok(Self { + reader: TlsReadHalf { + shared: Arc::clone(&shared), + read_socket, + pending_incoming: Vec::new(), + }, + writer: TlsWriteHalf { shared }, + }) + } + + /// Splits this stream into independent read and write halves. + pub fn into_split(self) -> (TlsReadHalf, TlsWriteHalf) { + (self.reader, self.writer) + } +} + +impl Read for TlsStream { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.reader.read(buf) + } +} + +impl Write for TlsStream { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.writer.write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + self.writer.flush() + } +} + +impl Read for TlsReadHalf { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + if buf.is_empty() { + return Ok(0); + } + loop { + if let Some(n) = self.try_read_plaintext(buf)? { + return Ok(n); + } + if self.pending_incoming.is_empty() && !self.read_ciphertext()? { + self.process_pending_incoming()?; + return match self.try_read_plaintext(buf)? { + Some(n) => Ok(n), + None => Err(io::Error::from(io::ErrorKind::UnexpectedEof)), + }; + } + self.process_pending_incoming()?; + } + } +} + +impl TlsReadHalf { + /// Reads buffered plaintext without touching the socket. + /// `None` needs more ciphertext; `Some(0)` is clean TLS EOF. + fn try_read_plaintext(&mut self, buf: &mut [u8]) -> io::Result> { + let mut conn = self.shared.tls_connection()?; + match conn.reader().read(buf) { + Err(err) if err.kind() == io::ErrorKind::WouldBlock => Ok(None), + other => other.map(Some), + } + } + + /// Reads TLS ciphertext into `pending_incoming`; returns `false` on TCP EOF. + fn read_ciphertext(&mut self) -> io::Result { + let mut buf = [0u8; CIPHERTEXT_CHUNK]; + loop { + match self.read_socket.read(&mut buf) { + Ok(0) => return Ok(false), + Ok(n) => { + self.pending_incoming.extend_from_slice(&buf[..n]); + return Ok(true); + } + Err(err) if err.kind() == io::ErrorKind::Interrupted => continue, + Err(err) => return Err(err), + } + } + } + + /// Feeds `pending_incoming` (or TCP EOF) into rustls and sends generated TLS responses. + fn process_pending_incoming(&mut self) -> io::Result<()> { + let mut protocol_err = None; + let wants_write = { + let mut conn = self.shared.tls_connection()?; + let n = { + let mut input = self.pending_incoming.as_slice(); + conn.read_tls(&mut input)? + }; + if n == 0 { + self.pending_incoming.clear(); + } else { + self.pending_incoming.drain(..n); + } + match conn.process_new_packets() { + Ok(_) => {} + Err(err) => protocol_err = Some(err), + } + // Make any queued TLS 1.3 KeyUpdate response available to `write_tls` + // without sending application data. + let _ = conn.writer().write(&[])?; + conn.wants_write() + }; + if wants_write { + let emitted = self.shared.emit(); + if protocol_err.is_none() { + emitted?; + } + } + match protocol_err { + Some(err) => Err(io::Error::new(io::ErrorKind::InvalidData, err)), + None => Ok(()), + } + } +} + +impl Drop for TlsReadHalf { + fn drop(&mut self) { + let _ = self.shared.shutdown_socket.shutdown(Shutdown::Both); + } +} + +impl Write for TlsWriteHalf { + fn write(&mut self, buf: &[u8]) -> io::Result { + if buf.is_empty() { + return Ok(0); + } + let mut output = self.shared.tls_output()?; + let prior_ciphertext = { + let mut conn = self.shared.tls_connection()?; + let mut ciphertext = Vec::new(); + drain_tls(&mut conn, &mut ciphertext)?; + ciphertext + }; + output.queue_and_write_ciphertext(&prior_ciphertext)?; + + let (ciphertext, n) = { + let mut conn = self.shared.tls_connection()?; + let mut ciphertext = Vec::new(); + let n = conn.writer().write(buf)?; + drain_tls(&mut conn, &mut ciphertext)?; + (ciphertext, n) + }; + // Plaintext accepted by rustls must be reported as written. + match output.queue_and_write_ciphertext(&ciphertext) { + Ok(()) => Ok(n), + Err(_err) if n > 0 => Ok(n), + Err(err) => Err(err), + } + } + + fn flush(&mut self) -> io::Result<()> { + self.shared.emit() + } +} + +impl Drop for TlsWriteHalf { + fn drop(&mut self) { + let _ = self.shared.shutdown_socket.shutdown(Shutdown::Both); + } +} + +/// Drains queued ciphertext so it can be written after releasing TLS state. +fn drain_tls(conn: &mut rustls::ClientConnection, out: &mut Vec) -> io::Result<()> { + while conn.wants_write() { + if conn.write_tls(out)? == 0 { + break; + } + } + Ok(()) +} From 415904976ab4ec6f49cb425ec01d008e9b91896b Mon Sep 17 00:00:00 2001 From: Noah Joeris Date: Wed, 26 Aug 2026 13:20:45 +0300 Subject: [PATCH 4/4] feat(client): add scheme-based connection facade Dispatch `tcp://` / `ssl://` URL strings to the TCP or TLS constructors for async and blocking clients. A missing scheme defaults to TCP. --- Cargo.lock | 352 ------------------------------------------ Cargo.toml | 2 - README.md | 33 ++-- src/client.rs | 138 +++++++++++++---- src/lib.rs | 4 +- src/transport/mod.rs | 166 ++++++++++++++++++-- tests/synopsis.rs | 358 +++++++++++++++++++++---------------------- 7 files changed, 455 insertions(+), 598 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 749f5de..4d6c234 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -40,125 +40,6 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" -[[package]] -name = "async-channel" -version = "1.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "81953c529336010edd6d8e358f886d9581267795c61b19475b71314bffa46d35" -dependencies = [ - "concurrent-queue", - "event-listener 2.5.3", - "futures-core", -] - -[[package]] -name = "async-channel" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2" -dependencies = [ - "concurrent-queue", - "event-listener-strategy", - "futures-core", - "pin-project-lite", -] - -[[package]] -name = "async-executor" -version = "1.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c96bf972d85afc50bf5ab8fe2d54d1586b4e0b46c97c50a0c9e71e2f7bcd812a" -dependencies = [ - "async-task", - "concurrent-queue", - "fastrand", - "futures-lite", - "pin-project-lite", - "slab", -] - -[[package]] -name = "async-global-executor" -version = "2.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05b1b633a2115cd122d73b955eadd9916c18c8f510ec9cd1686404c60ad1c29c" -dependencies = [ - "async-channel 2.5.0", - "async-executor", - "async-io", - "async-lock", - "blocking", - "futures-lite", - "once_cell", -] - -[[package]] -name = "async-io" -version = "2.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc" -dependencies = [ - "autocfg", - "cfg-if", - "concurrent-queue", - "futures-io", - "futures-lite", - "parking", - "polling", - "rustix 1.0.7", - "slab", - "windows-sys 0.61.2", -] - -[[package]] -name = "async-lock" -version = "3.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" -dependencies = [ - "event-listener 5.4.2", - "event-listener-strategy", - "pin-project-lite", -] - -[[package]] -name = "async-std" -version = "1.13.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2c8e079a4ab67ae52b7403632e4618815d6db36d2a010cfe41b02c1b1578f93b" -dependencies = [ - "async-channel 1.9.0", - "async-global-executor", - "async-io", - "async-lock", - "crossbeam-utils", - "futures-channel", - "futures-core", - "futures-io", - "futures-lite", - "gloo-timers", - "kv-log-macro", - "log", - "memchr", - "once_cell", - "pin-project-lite", - "pin-utils", - "slab", - "wasm-bindgen-futures", -] - -[[package]] -name = "async-task" -version = "4.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" - -[[package]] -name = "atomic-waker" -version = "1.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" - [[package]] name = "autocfg" version = "1.4.0" @@ -378,25 +259,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "blocking" -version = "1.6.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e83f8d02be6967315521be875afa792a316e28d57b5a2d401897e2a7921b7f21" -dependencies = [ - "async-channel 2.5.0", - "async-task", - "futures-io", - "futures-lite", - "piper", -] - -[[package]] -name = "bumpalo" -version = "3.20.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" - [[package]] name = "byteorder" version = "1.5.0" @@ -466,15 +328,6 @@ dependencies = [ "cc", ] -[[package]] -name = "concurrent-queue" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "constant_time_eq" version = "0.1.5" @@ -580,7 +433,6 @@ name = "electrum_streaming_client" version = "0.4.0" dependencies = [ "anyhow", - "async-std", "bdk_testenv", "bitcoin", "futures", @@ -603,32 +455,6 @@ dependencies = [ "windows-sys 0.59.0", ] -[[package]] -name = "event-listener" -version = "2.5.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" - -[[package]] -name = "event-listener" -version = "5.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" -dependencies = [ - "parking", - "pin-project-lite", -] - -[[package]] -name = "event-listener-strategy" -version = "0.5.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" -dependencies = [ - "event-listener 5.4.2", - "pin-project-lite", -] - [[package]] name = "fastrand" version = "2.3.0" @@ -709,7 +535,6 @@ dependencies = [ "futures-core", "futures-task", "futures-util", - "num_cpus", ] [[package]] @@ -718,19 +543,6 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" -[[package]] -name = "futures-lite" -version = "2.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" -dependencies = [ - "fastrand", - "futures-core", - "futures-io", - "parking", - "pin-project-lite", -] - [[package]] name = "futures-macro" version = "0.3.31" @@ -811,24 +623,6 @@ version = "0.31.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f" -[[package]] -name = "gloo-timers" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbb143cf96099802033e0d4f4963b19fd2e0b728bcf076cd9cf7f6634f092994" -dependencies = [ - "futures-channel", - "futures-core", - "js-sys", - "wasm-bindgen", -] - -[[package]] -name = "hermit-abi" -version = "0.5.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" - [[package]] name = "hex-conservative" version = "0.2.1" @@ -887,17 +681,6 @@ dependencies = [ "libc", ] -[[package]] -name = "js-sys" -version = "0.3.103" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" -dependencies = [ - "cfg-if", - "futures-util", - "wasm-bindgen", -] - [[package]] name = "jsonrpc" version = "0.18.0" @@ -910,15 +693,6 @@ dependencies = [ "serde_json", ] -[[package]] -name = "kv-log-macro" -version = "1.0.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0de8b303297635ad57c9f5059fd9cee7a47f8e8daa09df0fcd07dd39fb22977f" -dependencies = [ - "log", -] - [[package]] name = "libc" version = "0.2.172" @@ -963,9 +737,6 @@ name = "log" version = "0.4.27" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94" -dependencies = [ - "value-bag", -] [[package]] name = "memchr" @@ -1046,16 +817,6 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9" -[[package]] -name = "num_cpus" -version = "1.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" -dependencies = [ - "hermit-abi", - "libc", -] - [[package]] name = "object" version = "0.36.7" @@ -1071,12 +832,6 @@ version = "1.21.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" -[[package]] -name = "parking" -version = "2.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" - [[package]] name = "parking_lot" version = "0.12.4" @@ -1135,37 +890,12 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" -[[package]] -name = "piper" -version = "0.2.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" -dependencies = [ - "atomic-waker", - "fastrand", - "futures-io", -] - [[package]] name = "pkg-config" version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" -[[package]] -name = "polling" -version = "3.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" -dependencies = [ - "cfg-if", - "concurrent-queue", - "hermit-abi", - "pin-project-lite", - "rustix 1.0.7", - "windows-sys 0.61.2", -] - [[package]] name = "powerfmt" version = "0.2.0" @@ -1348,12 +1078,6 @@ dependencies = [ "untrusted", ] -[[package]] -name = "rustversion" -version = "1.0.23" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" - [[package]] name = "ryu" version = "1.0.20" @@ -1643,12 +1367,6 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" -[[package]] -name = "value-bag" -version = "1.13.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "068e763e8279de7ab94b6afebded2cb701678af094feb1c12ccb061b4783c1be" - [[package]] name = "version_check" version = "0.9.5" @@ -1670,61 +1388,6 @@ dependencies = [ "wit-bindgen-rt", ] -[[package]] -name = "wasm-bindgen" -version = "0.2.126" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" -dependencies = [ - "cfg-if", - "once_cell", - "rustversion", - "wasm-bindgen-macro", - "wasm-bindgen-shared", -] - -[[package]] -name = "wasm-bindgen-futures" -version = "0.4.76" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d" -dependencies = [ - "js-sys", - "wasm-bindgen", -] - -[[package]] -name = "wasm-bindgen-macro" -version = "0.2.126" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" -dependencies = [ - "quote", - "wasm-bindgen-macro-support", -] - -[[package]] -name = "wasm-bindgen-macro-support" -version = "0.2.126" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" -dependencies = [ - "bumpalo", - "proc-macro2", - "quote", - "syn 2.0.101", - "wasm-bindgen-shared", -] - -[[package]] -name = "wasm-bindgen-shared" -version = "0.2.126" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" -dependencies = [ - "unicode-ident", -] - [[package]] name = "webpki-roots" version = "0.25.4" @@ -1752,12 +1415,6 @@ dependencies = [ "rustix 0.38.44", ] -[[package]] -name = "windows-link" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" - [[package]] name = "windows-sys" version = "0.52.0" @@ -1776,15 +1433,6 @@ dependencies = [ "windows-targets", ] -[[package]] -name = "windows-sys" -version = "0.61.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" -dependencies = [ - "windows-link", -] - [[package]] name = "windows-targets" version = "0.52.6" diff --git a/Cargo.toml b/Cargo.toml index 74a9186..6189a5a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,8 +30,6 @@ tokio = ["dep:tokio", "tokio-util"] ssl = ["dep:rustls", "dep:webpki-roots", "dep:tokio-rustls"] [dev-dependencies] -async-std = "1.13.0" bdk_testenv = "0.11" -futures = { version = "0.3", features = ["thread-pool"] } tokio = { version = "1.44.2", features = ["full"] } anyhow = "1.0.98" diff --git a/README.md b/README.md index c4742ae..4e780d7 100644 --- a/README.md +++ b/README.md @@ -16,36 +16,39 @@ models. ## Example (async with Tokio) ```rust,no_run -use std::time::Duration; - -use electrum_streaming_client::{AsyncClient, ServerAddr}; +# #[cfg(all(feature = "tokio", feature = "ssl"))] +# mod example { +use electrum_streaming_client::{request, AsyncClient, ConnectConfig}; use futures::StreamExt; #[tokio::main] async fn main() -> anyhow::Result<()> { - let addr: ServerAddr = "127.0.0.1:50001".parse()?; - let (client, mut events, worker) = - AsyncClient::connect_tcp(&addr, Some(Duration::from_secs(10))).await?; - - tokio::spawn(worker); // spawn the client worker task - - let relay_fee = client.send_request(electrum_streaming_client::request::RelayFee).await?; + let (client, mut events, worker) = AsyncClient::connect( + "ssl://electrum.blockstream.info:50002", + &ConnectConfig::default(), + ) + .await?; + let worker = tokio::spawn(worker); + + let relay_fee = client.send_request(request::RelayFee).await?; println!("Relay fee: {relay_fee:?}"); - while let Some(event) = events.next().await { - println!("Event: {event:?}"); - } + client.send_event_request(request::HeadersSubscribe)?; + println!("Event: {:?}", events.next().await); + + drop(client); + worker.await??; Ok(()) } +# } ``` ## Optional Features -- `tokio`: Enables [`AsyncClient::new_tokio`] and [`AsyncClient::connect_tcp`]. +- `tokio` (default): Enables Tokio transport support. - `ssl`: Enables TLS via rustls. Async TLS additionally requires `tokio`. ## License MIT - diff --git a/src/client.rs b/src/client.rs index 63ed26f..08d7a10 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,4 +1,7 @@ +use std::{future::Future, io, net::Shutdown, net::TcpStream, thread::JoinHandle, time::Duration}; + use crate::pending_request::{PendingRequest, RequestExt}; +use crate::transport::ConnectTarget; use crate::*; // --- Async client type aliases --- @@ -94,7 +97,7 @@ impl AsyncClient { ) -> ( Self, AsyncEventReceiver, - impl std::future::Future> + Send, + impl Future> + Send, ) where R: futures::AsyncRead + Send + Unpin, @@ -123,7 +126,7 @@ impl AsyncClient { Some(incoming_res) => { let event_opt = state .handle_incoming(incoming_res?) - .map_err(|error| std::io::Error::new(std::io::ErrorKind::Other, error))?; + .map_err(|error| io::Error::new(io::ErrorKind::Other, error))?; if let Some(event) = event_opt { if let Err(_err) = event_tx.unbounded_send(event) { break; @@ -134,7 +137,7 @@ impl AsyncClient { } } } - std::io::Result::<()>::Ok(()) + io::Result::<()>::Ok(()) }; (Self { tx: req_tx }, event_recv, fut) @@ -164,7 +167,7 @@ impl AsyncClient { ) -> ( Self, AsyncEventReceiver, - impl std::future::Future> + Send, + impl Future> + Send, ) where R: tokio::io::AsyncRead + Send + Unpin, @@ -183,11 +186,11 @@ impl AsyncClient { #[cfg(feature = "tokio")] pub async fn connect_tcp( addr: &crate::transport::ServerAddr, - timeout: Option, - ) -> std::io::Result<( + timeout: Option, + ) -> io::Result<( Self, AsyncEventReceiver, - impl std::future::Future> + Send, + impl Future> + Send, )> { let stream = crate::transport::tokio::connect_tcp(addr, timeout).await?; let (reader, writer) = tokio::io::split(stream); @@ -202,12 +205,12 @@ impl AsyncClient { pub async fn connect_ssl( addr: &crate::transport::ServerAddr, validate_domain: bool, - timeout: Option, + timeout: Option, ) -> Result< ( Self, AsyncEventReceiver, - impl std::future::Future> + Send, + impl Future> + Send, ), crate::ConnectError, > { @@ -216,6 +219,47 @@ impl AsyncClient { Ok(Self::new_tokio(reader, writer)) } + /// Connects to `url` using its scheme. + /// + /// Accepts `host:port`, `tcp://host:port`, or `ssl://host:port`. + /// A missing scheme defaults to plaintext TCP; `ssl://` requires the `ssl` feature. + #[cfg(feature = "tokio")] + pub async fn connect( + url: &str, + config: &crate::transport::ConnectConfig, + ) -> Result< + ( + Self, + AsyncEventReceiver, + impl Future> + Send, + ), + crate::ConnectError, + > { + fn box_worker( + worker: F, + ) -> std::pin::Pin> + Send>> + where + F: Future> + Send + 'static, + { + Box::pin(worker) + } + + match url.parse::()? { + ConnectTarget::Tcp(addr) => { + let (client, events, worker) = Self::connect_tcp(&addr, config.timeout()).await?; + Ok((client, events, box_worker(worker))) + } + #[cfg(feature = "ssl")] + ConnectTarget::Ssl(addr) => { + let (client, events, worker) = + Self::connect_ssl(&addr, config.validate_domain(), config.timeout()).await?; + Ok((client, events, box_worker(worker))) + } + #[cfg(not(feature = "ssl"))] + ConnectTarget::Ssl(_) => Err(crate::ConnectError::SslUnsupported), + } + } + /// Sends a single tracked request to the Electrum server and awaits the response. /// /// This method is for request–response style interactions where only a single result is @@ -293,28 +337,28 @@ impl AsyncClient { } } -/// A `Write` wrapper around a [`std::net::TcpStream`] that shuts down the socket on drop. +/// A `Write` wrapper around a [`TcpStream`] that shuts down the socket on drop. /// /// [`BlockingClient::connect_tcp`] reads on a `try_clone` handle while the write thread holds /// this wrapper. When the last client handle drops, the write thread ends and dropping this /// wrapper shuts the socket down, unblocking the read thread. A plain `try_clone` handle would /// otherwise keep the socket alive until the peer closes it. #[derive(Debug)] -struct ShutdownOnDropTcpWriter(std::net::TcpStream); +struct ShutdownOnDropTcpWriter(TcpStream); -impl std::io::Write for ShutdownOnDropTcpWriter { - fn write(&mut self, buf: &[u8]) -> std::io::Result { - std::io::Write::write(&mut self.0, buf) +impl io::Write for ShutdownOnDropTcpWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.0.write(buf) } - fn flush(&mut self) -> std::io::Result<()> { - std::io::Write::flush(&mut self.0) + fn flush(&mut self) -> io::Result<()> { + self.0.flush() } } impl Drop for ShutdownOnDropTcpWriter { fn drop(&mut self) { - let _ = self.0.shutdown(std::net::Shutdown::Both); + let _ = self.0.shutdown(Shutdown::Both); } } @@ -366,27 +410,27 @@ impl BlockingClient { ) -> ( Self, BlockingEventReceiver, - std::thread::JoinHandle>, - std::thread::JoinHandle>, + JoinHandle>, + JoinHandle>, ) where - R: std::io::Read + Send + 'static, - W: std::io::Write + Send + 'static, + R: io::Read + Send + 'static, + W: io::Write + Send + 'static, { use std::sync::mpsc::*; let (event_tx, event_recv) = channel::(); let (req_tx, req_recv) = channel::>(); - let incoming_stream = crate::io::ReadStreamer::new(std::io::BufReader::new(reader)); + let incoming_stream = crate::io::ReadStreamer::new(io::BufReader::new(reader)); let read_state = std::sync::Arc::new(std::sync::Mutex::new(RequestTracker::new())); let write_state = std::sync::Arc::clone(&read_state); - let read_join = std::thread::spawn(move || -> std::io::Result<()> { + let read_join = std::thread::spawn(move || -> io::Result<()> { for incoming_res in incoming_stream { let event_opt = read_state .lock() .unwrap() .handle_incoming(incoming_res?) - .map_err(|error| std::io::Error::new(std::io::ErrorKind::Other, error))?; + .map_err(|error| io::Error::new(io::ErrorKind::Other, error))?; if let Some(event) = event_opt { if let Err(_err) = event_tx.send(event) { break; @@ -395,7 +439,7 @@ impl BlockingClient { } Ok(()) }); - let write_join = std::thread::spawn(move || -> std::io::Result<()> { + let write_join = std::thread::spawn(move || -> io::Result<()> { let mut next_id = 0_u32; for req in req_recv { let raw_req = write_state.lock().unwrap().track_request(&mut next_id, req); @@ -406,16 +450,46 @@ impl BlockingClient { (Self { tx: req_tx }, event_recv, read_join, write_join) } + /// Connects to `url` using its scheme. + /// + /// Accepts `host:port`, `tcp://host:port`, or `ssl://host:port`. + /// A missing scheme defaults to plaintext TCP; `ssl://` requires the `ssl` feature. + #[allow(clippy::type_complexity)] + pub fn connect( + url: &str, + config: &crate::transport::ConnectConfig, + ) -> Result< + ( + Self, + BlockingEventReceiver, + JoinHandle>, + JoinHandle>, + ), + crate::ConnectError, + > { + match url.parse::()? { + ConnectTarget::Tcp(addr) => { + Self::connect_tcp(&addr, config.timeout()).map_err(Into::into) + } + #[cfg(feature = "ssl")] + ConnectTarget::Ssl(addr) => { + Self::connect_ssl(&addr, config.validate_domain(), config.timeout()) + } + #[cfg(not(feature = "ssl"))] + ConnectTarget::Ssl(_) => Err(crate::ConnectError::SslUnsupported), + } + } + /// Creates a new [`BlockingClient`] connected to `addr` over plaintext TCP. #[allow(clippy::type_complexity)] pub fn connect_tcp( addr: &crate::transport::ServerAddr, - timeout: Option, - ) -> std::io::Result<( + timeout: Option, + ) -> io::Result<( Self, BlockingEventReceiver, - std::thread::JoinHandle>, - std::thread::JoinHandle>, + JoinHandle>, + JoinHandle>, )> { let stream = crate::transport::blocking::connect_tcp(addr, timeout)?; let reader = stream.try_clone()?; @@ -431,13 +505,13 @@ impl BlockingClient { pub fn connect_ssl( addr: &crate::transport::ServerAddr, validate_domain: bool, - timeout: Option, + timeout: Option, ) -> Result< ( Self, BlockingEventReceiver, - std::thread::JoinHandle>, - std::thread::JoinHandle>, + JoinHandle>, + JoinHandle>, ), crate::ConnectError, > { diff --git a/src/lib.rs b/src/lib.rs index 4ab719d..687224b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,7 +20,9 @@ pub use protocol::*; pub use request::Request; pub use request_tracker::*; pub use serde_json; -pub use transport::{ConnectError, Host, ParseServerAddrError, ServerAddr}; +pub use transport::{ + ConnectConfig, ConnectConfigBuilder, ConnectError, Host, ParseServerAddrError, ServerAddr, +}; #[cfg(feature = "ssl")] pub use transport::TlsError; diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 0a66010..5ae7e13 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -1,9 +1,7 @@ -//! Address types and TCP/TLS constructors for Electrum connections. +//! Address types and transport constructors for Electrum connections. //! -//! [`ServerAddr`] keeps the hostname unresolved so TLS SNI and certificate validation can use -//! the original name. Use [`blocking::connect_tcp`] / [`blocking::connect_ssl`] or the tokio -//! equivalents, or the client wrappers [`crate::BlockingClient::connect_tcp`] / -//! [`crate::AsyncClient::connect_tcp`]. +//! Use [`crate::BlockingClient::connect`] or [`crate::AsyncClient::connect`] for built-in +//! TCP/TLS, or [`crate::BlockingClient::new`] / [`crate::AsyncClient::new`] for custom I/O. use std::fmt; use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; @@ -14,6 +12,36 @@ use std::sync::Arc; #[cfg(feature = "ssl")] mod tls; +/// A parsed Electrum server address and transport. +/// +/// Accepts `[tcp://|ssl://]host:port`; no scheme defaults to TCP. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum ConnectTarget { + /// Plaintext TCP (`tcp://` or no scheme prefix). + Tcp(ServerAddr), + /// SSL/TLS encrypted TCP (`ssl://`). + #[cfg_attr(not(feature = "ssl"), allow(dead_code))] + Ssl(ServerAddr), +} + +impl FromStr for ConnectTarget { + type Err = ParseServerAddrError; + + fn from_str(s: &str) -> Result { + let parse_addr = |addr: &str| { + addr.parse::() + .map_err(|_| ParseServerAddrError(s.to_string())) + }; + + match s.split_once("://") { + Some(("tcp", addr)) => Ok(Self::Tcp(parse_addr(addr)?)), + Some(("ssl", addr)) => Ok(Self::Ssl(parse_addr(addr)?)), + Some(_) => Err(ParseServerAddrError(s.to_string())), + None => Ok(Self::Tcp(parse_addr(s)?)), + } + } +} + /// The host portion of a [`ServerAddr`]. #[derive(Debug, Clone, PartialEq, Eq)] pub enum Host { @@ -36,6 +64,8 @@ impl fmt::Display for Host { /// An Electrum server address: a [`Host`] and a port, without a connection scheme. /// /// Parses from `"host:port"`. IPv6 literals must be bracketed (`"[::1]:50001"`). +/// Scheme prefixes are rejected; use [`crate::BlockingClient::connect`] or +/// [`crate::AsyncClient::connect`] for scheme-prefixed input. #[derive(Debug, Clone, PartialEq, Eq)] pub struct ServerAddr { host: Host, @@ -113,7 +143,7 @@ impl ToSocketAddrs for ServerAddr { } } -/// An error parsing a [`ServerAddr`] from a string. +/// An error parsing a [`ServerAddr`]. /// /// The payload is the full input string that failed to parse. #[derive(Debug, Clone, PartialEq, Eq)] @@ -127,15 +157,87 @@ impl fmt::Display for ParseServerAddrError { impl std::error::Error for ParseServerAddrError {} +/// Configuration for [`crate::client::BlockingClient::connect`] / +/// [`crate::client::AsyncClient::connect`]. +/// +/// Use [`ConnectConfig::builder`] to construct. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ConnectConfig { + /// Timeout for establishing the connection (`None` = no limit). + timeout: Option, + /// Whether to validate the server's TLS certificate against the domain (TLS only). + validate_domain: bool, +} + +impl ConnectConfig { + /// Returns a [`ConnectConfigBuilder`] with default values. + pub fn builder() -> ConnectConfigBuilder { + ConnectConfigBuilder::default() + } + + /// Timeout for establishing the connection. + /// + /// `None` means no limit. + pub fn timeout(&self) -> Option { + self.timeout + } + + /// Whether to validate the server's TLS certificate against the domain. + /// + /// This only applies to TLS connections and is ignored for plain TCP. Defaults to `true`. + pub fn validate_domain(&self) -> bool { + self.validate_domain + } +} + +impl Default for ConnectConfig { + fn default() -> Self { + Self { + timeout: None, + validate_domain: true, + } + } +} + +/// A builder for [`ConnectConfig`], obtained via [`ConnectConfig::builder`]. +#[derive(Debug, Clone, Default)] +pub struct ConnectConfigBuilder { + config: ConnectConfig, +} + +impl ConnectConfigBuilder { + /// Sets the connection timeout. See [`ConnectConfig::timeout`]. + pub fn timeout(mut self, timeout: Option) -> Self { + self.config.timeout = timeout; + self + } + + /// Sets whether to validate the server's TLS certificate against the domain. See + /// [`ConnectConfig::validate_domain`]. + pub fn validate_domain(mut self, validate_domain: bool) -> Self { + self.config.validate_domain = validate_domain; + self + } + + /// Builds the [`ConnectConfig`]. + pub fn build(self) -> ConnectConfig { + self.config + } +} + /// Error establishing an Electrum connection. #[non_exhaustive] #[derive(Debug)] pub enum ConnectError { + /// The server address failed to parse. + InvalidServerAddr(ParseServerAddrError), /// Transport or socket failure, including DNS, TCP, timeout, and handshake EOF/reset. Io(std::io::Error), /// TLS configuration, protocol, or certificate-validation failure. #[cfg(feature = "ssl")] Tls(TlsError), + /// The URL uses `ssl://` but the crate was built without the `ssl` feature. + SslUnsupported, } #[cfg(feature = "ssl")] @@ -163,9 +265,13 @@ impl ConnectError { impl fmt::Display for ConnectError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { + ConnectError::InvalidServerAddr(error) => write!(f, "{error}"), ConnectError::Io(e) => write!(f, "connection I/O error: {e}"), #[cfg(feature = "ssl")] ConnectError::Tls(e) => write!(f, "{e}"), + ConnectError::SslUnsupported => { + write!(f, "the 'ssl://' scheme requires the 'ssl' feature") + } } } } @@ -173,9 +279,11 @@ impl fmt::Display for ConnectError { impl std::error::Error for ConnectError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { + ConnectError::InvalidServerAddr(error) => Some(error), ConnectError::Io(e) => Some(e), #[cfg(feature = "ssl")] ConnectError::Tls(e) => Some(e), + ConnectError::SslUnsupported => None, } } } @@ -186,6 +294,12 @@ impl From for ConnectError { } } +impl From for ConnectError { + fn from(error: ParseServerAddrError) -> Self { + ConnectError::InvalidServerAddr(error) + } +} + #[cfg(feature = "ssl")] impl From for ConnectError { fn from(e: TlsError) -> Self { @@ -586,22 +700,18 @@ mod tests { use super::*; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; - fn addr(s: &str) -> ServerAddr { - s.parse().unwrap_or_else(|e| panic!("{s:?}: {e}")) - } - #[test] fn server_addr_parse() { - let a = addr("127.0.0.1:50001"); + let a: ServerAddr = "127.0.0.1:50001".parse().unwrap(); assert_eq!(a.host(), &Host::Ip(IpAddr::V4(Ipv4Addr::LOCALHOST))); assert_eq!(a.port(), 50001); assert_eq!(a.to_string(), "127.0.0.1:50001"); - let a = addr("localhost:50001"); + let a: ServerAddr = "localhost:50001".parse().unwrap(); assert_eq!(a.host(), &Host::Domain("localhost".into())); assert_eq!(a.port(), 50001); - let a = addr("[::1]:50001"); + let a: ServerAddr = "[::1]:50001".parse().unwrap(); assert_eq!(a.host(), &Host::Ip(IpAddr::V6(Ipv6Addr::LOCALHOST))); assert_eq!(a.port(), 50001); assert_eq!(a.to_string(), "[::1]:50001"); @@ -620,4 +730,34 @@ mod tests { assert!(bad.parse::().is_err(), "{bad}"); } } + + #[test] + fn connect_target_parse() { + assert_eq!( + "127.0.0.1:50001".parse::().unwrap(), + ConnectTarget::Tcp("127.0.0.1:50001".parse().unwrap()) + ); + assert_eq!( + "tcp://electrum.example.com:50001" + .parse::() + .unwrap(), + ConnectTarget::Tcp("electrum.example.com:50001".parse().unwrap()) + ); + assert_eq!( + "ssl://[::1]:50002".parse::().unwrap(), + ConnectTarget::Ssl("[::1]:50002".parse().unwrap()) + ); + + for bad in ["http://host:80", "TCP://host:1", "ssl://", "tcp://host"] { + assert!(bad.parse::().is_err(), "{bad}"); + } + } + + #[test] + fn client_connect_rejects_invalid_server_addr() { + assert!(matches!( + crate::BlockingClient::connect("http://host:80", &ConnectConfig::default()), + Err(ConnectError::InvalidServerAddr(error)) if error.0 == "http://host:80" + )); + } } diff --git a/tests/synopsis.rs b/tests/synopsis.rs index 129a605..013d0c3 100644 --- a/tests/synopsis.rs +++ b/tests/synopsis.rs @@ -1,206 +1,198 @@ +#![cfg(feature = "tokio")] + use std::time::Duration; -use async_std::{net::TcpStream, stream::StreamExt}; use bdk_testenv::{anyhow, bitcoincore_rpc::RpcApi, TestEnv}; use bitcoin::Amount; use electrum_streaming_client::{ - notification::Notification, request, AsyncClient, CompletedRequest, Event, -}; -use futures::{ - executor::{block_on, ThreadPool}, - task::SpawnExt, - AsyncReadExt, FutureExt, + notification::Notification, request, AsyncClient, CompletedRequest, ConnectConfig, Event, }; +use futures::{FutureExt, StreamExt}; -#[test] -fn synopsis() -> anyhow::Result<()> { +#[tokio::test] +async fn synopsis() -> anyhow::Result<()> { let env = TestEnv::new()?; - let electrum_addr = env.electrsd.electrum_url.clone(); - println!("URL: {}", electrum_addr); + let url = format!("tcp://{}", env.electrsd.electrum_url); let wallet_addr = env .rpc_client() .get_new_address(None, None)? .assume_checked(); - let pool = ThreadPool::new()?; - block_on(async { - let stream = TcpStream::connect(electrum_addr.as_str()).await?; - let (read_stream, write_strean) = stream.split(); - let (client, mut event_rx, run_fut) = AsyncClient::new(read_stream, write_strean); - let run_handle = pool.spawn_with_handle(run_fut)?; - - client.send_event_request(request::HeadersSubscribe)?; - client.send_event_request(request::ScriptHashSubscribe::from_script( - wallet_addr.script_pubkey(), - ))?; - assert!(matches!( - event_rx.next().await, - Some(Event::Response(CompletedRequest::HeadersSubscribe { .. })) - )); - assert!(matches!( - event_rx.next().await, - Some(Event::Response( - CompletedRequest::ScriptHashSubscribe { .. } - )) - )); - - const TO_MINE: usize = 3; - let blockhashes = env.mine_blocks(TO_MINE, Some(wallet_addr.clone()))?; - println!("MINED: {:?}", blockhashes); - while let Some(event) = event_rx.next().await { - if let Event::Notification(Notification::Header(n)) = event { - if n.height() > TO_MINE as u32 { - break; - } + let (client, mut event_rx, run_fut) = + AsyncClient::connect(&url, &ConnectConfig::default()).await?; + let run_handle = tokio::spawn(run_fut); + + client.send_event_request(request::HeadersSubscribe)?; + client.send_event_request(request::ScriptHashSubscribe::from_script( + wallet_addr.script_pubkey(), + ))?; + assert!(matches!( + event_rx.next().await, + Some(Event::Response(CompletedRequest::HeadersSubscribe { .. })) + )); + assert!(matches!( + event_rx.next().await, + Some(Event::Response( + CompletedRequest::ScriptHashSubscribe { .. } + )) + )); + + const TO_MINE: usize = 3; + let blockhashes = env.mine_blocks(TO_MINE, Some(wallet_addr.clone()))?; + println!("MINED: {:?}", blockhashes); + while let Some(event) = event_rx.next().await { + if let Event::Notification(Notification::Header(n)) = event { + if n.height() > TO_MINE as u32 { + break; } } + } - assert_eq!( - client - .send_request(request::HeaderWithProof { - height: 3, - cp_height: 3 - }) - .await? - .header, - { - let blockhash = env.rpc_client().get_block_hash(3)?; - env.rpc_client().get_block_header(&blockhash)? - }, - "header at height must match" - ); - - println!( - "HEADERS: {:?}", - client - .send_request(request::Headers { - start_height: 1, - count: 2, - }) - .await? - ); - - // Make unconfirmed balance. - env.mine_blocks(101, Some(wallet_addr.clone()))?; // create spendable balance - let txid = env.rpc_client().send_to_address( - &wallet_addr, - Amount::from_btc(1.0).unwrap(), - None, - None, - None, - None, - None, - None, - )?; - env.wait_until_electrum_sees_txid(txid, Duration::from_secs(10))?; - - let tx_resp = client.send_request(request::GetTx { txid }).await?; - println!("GOT TX: {:?}", tx_resp); - println!( - "BROADCAST RESULT: {}", - client - .send_request(request::BroadcastTx(tx_resp.tx)) - .await? - ); - - println!( - "GET BALANCE RESP: {:?}", - client - .send_request(request::GetBalance::from_script( - wallet_addr.script_pubkey(), - )) - .await? - ); - - let history_resp = client - .send_request(request::GetHistory::from_script( - wallet_addr.script_pubkey(), - )) - .await?; - println!( - "GET HISTORY RESP: first = {:?} last = {:?}", - history_resp.first().unwrap(), - history_resp.last().unwrap() - ); - - let block_hash = env.mine_blocks(1, None)?.first().copied().unwrap(); - let block_height = env.rpc_client().get_block_info(&block_hash)?.height as u32; - env.wait_until_electrum_sees_block(Duration::from_secs(5))?; - - let tx_merkle = client - .send_request(request::GetTxMerkle { - txid, - height: block_height, + assert_eq!( + client + .send_request(request::HeaderWithProof { + height: 3, + cp_height: 3 }) - .await?; - println!("GET MERKLE: {:?}", tx_merkle); - - let from_pos = client - .send_request(request::GetTxidFromPos { - height: block_height, - tx_pos: tx_merkle.pos, + .await? + .header, + { + let blockhash = env.rpc_client().get_block_hash(3)?; + env.rpc_client().get_block_header(&blockhash)? + }, + "header at height must match" + ); + + println!( + "HEADERS: {:?}", + client + .send_request(request::Headers { + start_height: 1, + count: 2, }) - .await?; - println!("TXID FROM POS: {}", from_pos.txid); - assert_eq!(txid, from_pos.txid); - - // NOTE: This does not work with `electrs` - // let mempool_history = request::GetMempool::from_script(addr.script_pubkey()) - // .send(&req_tx)? - // .await??; - // println!("GET MEMPOOL RESP: {:?}", mempool_history); - - let utxos = client - .send_request(request::ListUnspent::from_script( + .await? + ); + + // Make unconfirmed balance. + env.mine_blocks(101, Some(wallet_addr.clone()))?; // create spendable balance + let txid = env.rpc_client().send_to_address( + &wallet_addr, + Amount::from_btc(1.0).unwrap(), + None, + None, + None, + None, + None, + None, + )?; + env.wait_until_electrum_sees_txid(txid, Duration::from_secs(10))?; + + let tx_resp = client.send_request(request::GetTx { txid }).await?; + println!("GOT TX: {:?}", tx_resp); + println!( + "BROADCAST RESULT: {}", + client + .send_request(request::BroadcastTx(tx_resp.tx)) + .await? + ); + + println!( + "GET BALANCE RESP: {:?}", + client + .send_request(request::GetBalance::from_script( wallet_addr.script_pubkey(), )) - .await?; - println!( - "GET UTXOs: first = {:?} last = {:?}", - utxos.first().unwrap(), - utxos.last().unwrap() - ); - - // NOTE: This does not work with our version of `electrs` - // let unsub_resp = request::ScriptHashUnsubscribe::from_script(addr.script_pubkey()) - // .send(&req_tx)? - // .await??; - // println!("UNSUB RESP: {:?}", unsub_resp); - - let fee_histogram = client.send_request(request::GetFeeHistogram).await?; - println!("FEE HISTOGRAM: {:?}", fee_histogram); - - let server_banner = client.send_request(request::Banner).await?; - println!("SERVER BANNER: {}", server_banner); - - client.send_request(request::Ping).await?; - println!("PING SUCCESS!"); - - // // NOTE: Batching does not work until https://github.com/Blockstream/electrs/pull/108 is - // // merged. - // - // let txid1 = env.send(&wallet_addr, Amount::from_btc(0.1)?)?; - // let txid2 = env.send(&wallet_addr, Amount::from_btc(0.1)?)?; - // env.mine_blocks(1, None)?; - // env.wait_until_electrum_sees_txid(txid1, Duration::from_secs(10))?; - // env.wait_until_electrum_sees_txid(txid2, Duration::from_secs(10))?; - // let mut batch = client.batch(); - // let tx1_fut = batch.request(request::GetTx(txid1)); - // let tx2_fut = batch.request(request::GetTx(txid2)); - // batch.send()?; - // let (tx1_res, tx2_res) = futures::join!(tx1_fut, tx2_fut); - // let (tx1, tx2) = (tx1_res?, tx2_res?); - // println!("Got tx1: {:?}", tx1); - // println!("Got tx2: {:?}", tx2); - - // read remaining events. - while let Some(event) = event_rx.next().now_or_never() { - println!("EVENT: {:?}", event); - } + .await? + ); - drop(client); - run_handle.await?; - Ok(()) - }) + let history_resp = client + .send_request(request::GetHistory::from_script( + wallet_addr.script_pubkey(), + )) + .await?; + println!( + "GET HISTORY RESP: first = {:?} last = {:?}", + history_resp.first().unwrap(), + history_resp.last().unwrap() + ); + + let block_hash = env.mine_blocks(1, None)?.first().copied().unwrap(); + let block_height = env.rpc_client().get_block_info(&block_hash)?.height as u32; + env.wait_until_electrum_sees_block(Duration::from_secs(5))?; + + let tx_merkle = client + .send_request(request::GetTxMerkle { + txid, + height: block_height, + }) + .await?; + println!("GET MERKLE: {:?}", tx_merkle); + + let from_pos = client + .send_request(request::GetTxidFromPos { + height: block_height, + tx_pos: tx_merkle.pos, + }) + .await?; + println!("TXID FROM POS: {}", from_pos.txid); + assert_eq!(txid, from_pos.txid); + + // NOTE: This does not work with `electrs` + // let mempool_history = request::GetMempool::from_script(addr.script_pubkey()) + // .send(&req_tx)? + // .await??; + // println!("GET MEMPOOL RESP: {:?}", mempool_history); + + let utxos = client + .send_request(request::ListUnspent::from_script( + wallet_addr.script_pubkey(), + )) + .await?; + println!( + "GET UTXOs: first = {:?} last = {:?}", + utxos.first().unwrap(), + utxos.last().unwrap() + ); + + // NOTE: This does not work with our version of `electrs` + // let unsub_resp = request::ScriptHashUnsubscribe::from_script(addr.script_pubkey()) + // .send(&req_tx)? + // .await??; + // println!("UNSUB RESP: {:?}", unsub_resp); + + let fee_histogram = client.send_request(request::GetFeeHistogram).await?; + println!("FEE HISTOGRAM: {:?}", fee_histogram); + + let server_banner = client.send_request(request::Banner).await?; + println!("SERVER BANNER: {}", server_banner); + + client.send_request(request::Ping).await?; + println!("PING SUCCESS!"); + + // // NOTE: Batching does not work until https://github.com/Blockstream/electrs/pull/108 is + // // merged. + // + // let txid1 = env.send(&wallet_addr, Amount::from_btc(0.1)?)?; + // let txid2 = env.send(&wallet_addr, Amount::from_btc(0.1)?)?; + // env.mine_blocks(1, None)?; + // env.wait_until_electrum_sees_txid(txid1, Duration::from_secs(10))?; + // env.wait_until_electrum_sees_txid(txid2, Duration::from_secs(10))?; + // let mut batch = client.batch(); + // let tx1_fut = batch.request(request::GetTx(txid1)); + // let tx2_fut = batch.request(request::GetTx(txid2)); + // batch.send()?; + // let (tx1_res, tx2_res) = futures::join!(tx1_fut, tx2_fut); + // let (tx1, tx2) = (tx1_res?, tx2_res?); + // println!("Got tx1: {:?}", tx1); + // println!("Got tx2: {:?}", tx2); + + // read remaining events. + while let Some(event) = event_rx.next().now_or_never() { + println!("EVENT: {:?}", event); + } + + drop(client); + run_handle.await??; + Ok(()) }