From 33cbf26d16ff658f77587b02c09b74ae89ff8763 Mon Sep 17 00:00:00 2001 From: Noah Joeris Date: Tue, 18 Aug 2026 23:32:19 +0300 Subject: [PATCH 1/2] feat: add transport address types and plaintext TCP connect Keep hostnames unresolved for TLS/SOCKS, and give blocking/tokio clients a connect-only TCP constructor. --- Cargo.toml | 2 +- README.md | 14 +- src/client.rs | 31 ++++ src/lib.rs | 5 + src/transport.rs | 463 +++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 507 insertions(+), 8 deletions(-) create mode 100644 src/transport.rs diff --git a/Cargo.toml b/Cargo.toml index 5e3c99a..554965b 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..dd2c33c 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,9 +42,8 @@ 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 MIT - diff --git a/src/client.rs b/src/client.rs index 2c55200..b06d5cc 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 @@ -344,6 +359,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 writer = crate::transport::blocking::connect_tcp(addr, timeout)?; + let reader = writer.try_clone()?; + 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/lib.rs b/src/lib.rs index c48897e..1cf49cf 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -13,12 +13,17 @@ 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::{ + ConnectConfig, ConnectConfigBuilder, Host, ParseServerAddrError, Scheme, ServerAddr, ServerUrl, + Socks5Config, Socks5Credential, +}; /// An owned or borrowed static string. pub type CowStr = std::borrow::Cow<'static, str>; diff --git a/src/transport.rs b/src/transport.rs new file mode 100644 index 0000000..3e6a683 --- /dev/null +++ b/src/transport.rs @@ -0,0 +1,463 @@ +//! Address and configuration types for Electrum server connections. +//! +//! The address types ([`Scheme`], [`Host`], [`ServerAddr`], [`ServerUrl`]) represent the target +//! server for the transport constructors. The hostname is kept as a [`Host::Domain`] (rather +//! than being eagerly resolved), which is required for: +//! +//! * TLS SNI and certificate validation, and +//! * SOCKS5 proxy connections, where DNS resolution must happen proxy-side (e.g. `.onion`). +//! +//! The configuration types ([`ConnectConfig`], [`ConnectConfigBuilder`], [`Socks5Config`], +//! [`Socks5Credential`]) carry connection options such as timeouts, TLS certificate validation +//! and an optional SOCKS5 proxy. +//! +//! # Parsing rules +//! +//! [`ServerUrl`] parses `"[scheme://]host:port"`: +//! +//! | Input | Result | +//! |---|---| +//! | `host:50001` | scheme defaults to [`Scheme::Tcp`] | +//! | `tcp://host:50001`, `ssl://host:50001` | the respective scheme | +//! | `foo://host:50001`, `SSL://host:50001` | [`ParseServerAddrError`] (lowercase only) | +//! | `127.0.0.1:50001`, `[::1]:50001` | [`Host::Ip`] (IPv6 requires brackets) | +//! | `::1:50001` | [`ParseServerAddrError`] (unbracketed IPv6) | +//! | `….onion:50001` | [`Host::Domain`] (no special-casing) | +//! | `""`, `ssl://`, `ssl://host`, `host:`, `:50001`, `host:abc`, `host:99999` | [`ParseServerAddrError`] | +//! | `host:0` | valid (port 0 is allowed) | +//! +//! [`ServerAddr`] parses the same `host:port` grammar, but rejects any input containing +//! `"://"`. +//! +//! Additional rules: +//! +//! * No trimming; no default ports. +//! * Domains are not validated beyond being non-empty; unresolvable names fail later at +//! connection time. +//! * The error payload is the full input string that failed to parse. +//! +//! # Connecting +//! +//! The actual transport constructors live in submodules: +//! +//! * [`blocking`]: blocking (std I/O) constructors, e.g. [`blocking::connect_tcp`]. +//! * [`tokio`]: Tokio-based async constructors, e.g. [`tokio::connect_tcp`] (feature `tokio`). + +use std::fmt; +use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; +use std::str::FromStr; + +/// The connection scheme of an Electrum server, taken from the URL prefix. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Scheme { + /// Plaintext TCP (`tcp://`). + Tcp, + /// SSL/TLS encrypted TCP (`ssl://`). + Ssl, +} + +impl Scheme { + /// The URL prefix of this scheme (e.g. `tcp` for `tcp://`). + pub fn as_str(&self) -> &'static str { + match self { + Scheme::Tcp => "tcp", + Scheme::Ssl => "ssl", + } + } +} + +impl fmt::Display for Scheme { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl FromStr for Scheme { + type Err = ParseServerAddrError; + + /// Parses a scheme prefix. Lowercase only: `"tcp"` and `"ssl"` are accepted. + fn from_str(s: &str) -> Result { + match s { + "tcp" => Ok(Scheme::Tcp), + "ssl" => Ok(Scheme::Ssl), + other => Err(ParseServerAddrError(other.to_string())), + } + } +} + +/// The host portion of a [`ServerAddr`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Host { + /// A domain name, e.g. `electrum.example.com` or `….onion`. + 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"`). +/// Use [`ServerUrl`] to parse scheme-prefixed addresses. +#[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` or other SOCKS5 targets — those must be passed to the + /// proxy as an unresolved domain (proxy-side 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(), + } + } +} + +/// A full Electrum server URL: a connection [`Scheme`] and a [`ServerAddr`]. +/// +/// Parses from `"[scheme://]host:port"`. A missing scheme defaults to [`Scheme::Tcp`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ServerUrl { + scheme: Scheme, + addr: ServerAddr, +} + +impl ServerUrl { + /// Creates a new `ServerUrl` from a [`Scheme`] and [`ServerAddr`]. + pub fn new(scheme: Scheme, addr: ServerAddr) -> Self { + Self { scheme, addr } + } + + /// The connection scheme of this URL. + pub fn scheme(&self) -> Scheme { + self.scheme + } + + /// The server address (host and port) of this URL. + pub fn addr(&self) -> &ServerAddr { + &self.addr + } +} + +impl fmt::Display for ServerUrl { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}://{}", self.scheme, self.addr) + } +} + +impl FromStr for ServerUrl { + type Err = ParseServerAddrError; + + fn from_str(s: &str) -> Result { + let invalid = || ParseServerAddrError(s.to_string()); + let (scheme, addr_str) = match s.split_once("://") { + Some((scheme_str, addr_str)) => (scheme_str.parse().map_err(|_| invalid())?, addr_str), + None => (Scheme::Tcp, s), + }; + Ok(Self { + scheme, + addr: addr_str.parse().map_err(|_| invalid())?, + }) + } +} + +/// An error parsing a [`Scheme`], [`ServerAddr`] or [`ServerUrl`] 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 {} + +/// Credential for a SOCKS5 proxy. +#[derive(Clone, PartialEq, Eq)] +pub struct Socks5Credential { + /// Username for SOCKS5 authentication. + pub username: String, + /// Password for SOCKS5 authentication. + pub password: String, +} + +impl fmt::Debug for Socks5Credential { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Socks5Credential") + .field("username", &self.username) + .field("password", &"") + .finish() + } +} + +/// Configuration of a SOCKS5 proxy, e.g. for connecting over Tor. +/// +/// The proxy address is typically a local address (e.g. `127.0.0.1:9050` for Tor). DNS +/// resolution of the *target* server happens proxy-side, which is what makes `.onion` +/// addresses reachable. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Socks5Config { + /// The address of the SOCKS5 proxy. + pub addr: ServerAddr, + /// Credentials used to authenticate with the proxy, if any. + pub credentials: Option, +} + +impl Socks5Config { + /// Creates a new `Socks5Config` for a proxy that requires no authentication. + pub fn new(addr: ServerAddr) -> Self { + Self { + addr, + credentials: None, + } + } + + /// Creates a new `Socks5Config` for a proxy that requires username/password + /// authentication. + pub fn with_credentials(addr: ServerAddr, username: String, password: String) -> Self { + Self { + addr, + credentials: Some(Socks5Credential { username, password }), + } + } +} + +/// Configuration for establishing a connection to an Electrum server. +/// +/// 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, + /// The SOCKS5 proxy to connect through, if any. + socks5: Option, +} + +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 + } + + /// The SOCKS5 proxy to connect through, if any. + pub fn socks5(&self) -> Option<&Socks5Config> { + self.socks5.as_ref() + } +} + +impl Default for ConnectConfig { + fn default() -> Self { + Self { + timeout: None, + validate_domain: true, + socks5: None, + } + } +} + +/// 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 + } + + /// Sets the SOCKS5 proxy to connect through. See [`ConnectConfig::socks5`]. + pub fn socks5(mut self, socks5: Option) -> Self { + self.config.socks5 = socks5; + self + } + + /// Builds the [`ConnectConfig`]. + pub fn build(self) -> ConnectConfig { + self.config + } +} + +/// 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, + } + } +} From 1ccff52ce8a38b5ca130087d5e9e9b626707126e Mon Sep 17 00:00:00 2001 From: Noah Joeris Date: Thu, 20 Aug 2026 23:21:12 +0300 Subject: [PATCH 2/2] feat(transport): add SSL transport --- Cargo.lock | 131 +++++++- Cargo.toml | 11 +- README.md | 1 + src/client.rs | 46 +++ src/io.rs | 9 +- src/lib.rs | 4 +- src/transport.rs | 463 -------------------------- src/transport/mod.rs | 748 +++++++++++++++++++++++++++++++++++++++++++ src/transport/tls.rs | 311 ++++++++++++++++++ 9 files changed, 1247 insertions(+), 477 deletions(-) delete mode 100644 src/transport.rs create mode 100644 src/transport/mod.rs create mode 100644 src/transport/tls.rs diff --git a/Cargo.lock b/Cargo.lock index 24b483e..d36bcda 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -166,6 +166,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" @@ -409,10 +432,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 +458,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 +536,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 +585,13 @@ dependencies = [ "bdk_testenv", "bitcoin", "futures", + "rustls 0.23.43", "serde", "serde_json", "tokio", + "tokio-rustls", "tokio-util", + "webpki-roots 1.0.9", ] [[package]] @@ -607,6 +649,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 +665,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" @@ -955,11 +1009,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]] @@ -1246,10 +1300,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,6 +1338,18 @@ 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" @@ -1365,9 +1455,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" @@ -1492,6 +1582,16 @@ dependencies = [ "syn", ] +[[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]] name = "tokio-util" version = "0.7.15" @@ -1654,6 +1754,15 @@ version = "0.25.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" +[[package]] +name = "webpki-roots" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "which" version = "4.4.2" @@ -1787,6 +1896,12 @@ dependencies = [ "syn", ] +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + [[package]] name = "zip" version = "0.6.6" diff --git a/Cargo.toml b/Cargo.toml index 554965b..74a9186 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,23 +4,30 @@ 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" +[package.metadata.docs.rs] +all-features = true + [dependencies] futures = "0.3" serde = { version = "1", features = ["derive"] } serde_json = "1" bitcoin = { version = "0.32", features = ["serde"] } -tokio = { version = "1.44.2", features = ["io-util", "net", "time"], optional = true } +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 dd2c33c..f4803a6 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 (`BlockingClient::connect_ssl`; `AsyncClient::connect_ssl`). ## License diff --git a/src/client.rs b/src/client.rs index b06d5cc..e5b0feb 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::TlsConnectError, + > { + 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 @@ -375,6 +397,30 @@ impl BlockingClient { Ok(Self::new(reader, writer)) } + /// 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::TlsConnectError, + > { + 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 1cf49cf..0f2f4b4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,9 +22,11 @@ pub use request_tracker::*; pub use serde_json; pub use transport::{ ConnectConfig, ConnectConfigBuilder, Host, ParseServerAddrError, Scheme, ServerAddr, ServerUrl, - Socks5Config, Socks5Credential, }; +#[cfg(feature = "ssl")] +pub use transport::TlsConnectError; + /// An owned or borrowed static string. pub type CowStr = std::borrow::Cow<'static, str>; diff --git a/src/transport.rs b/src/transport.rs deleted file mode 100644 index 3e6a683..0000000 --- a/src/transport.rs +++ /dev/null @@ -1,463 +0,0 @@ -//! Address and configuration types for Electrum server connections. -//! -//! The address types ([`Scheme`], [`Host`], [`ServerAddr`], [`ServerUrl`]) represent the target -//! server for the transport constructors. The hostname is kept as a [`Host::Domain`] (rather -//! than being eagerly resolved), which is required for: -//! -//! * TLS SNI and certificate validation, and -//! * SOCKS5 proxy connections, where DNS resolution must happen proxy-side (e.g. `.onion`). -//! -//! The configuration types ([`ConnectConfig`], [`ConnectConfigBuilder`], [`Socks5Config`], -//! [`Socks5Credential`]) carry connection options such as timeouts, TLS certificate validation -//! and an optional SOCKS5 proxy. -//! -//! # Parsing rules -//! -//! [`ServerUrl`] parses `"[scheme://]host:port"`: -//! -//! | Input | Result | -//! |---|---| -//! | `host:50001` | scheme defaults to [`Scheme::Tcp`] | -//! | `tcp://host:50001`, `ssl://host:50001` | the respective scheme | -//! | `foo://host:50001`, `SSL://host:50001` | [`ParseServerAddrError`] (lowercase only) | -//! | `127.0.0.1:50001`, `[::1]:50001` | [`Host::Ip`] (IPv6 requires brackets) | -//! | `::1:50001` | [`ParseServerAddrError`] (unbracketed IPv6) | -//! | `….onion:50001` | [`Host::Domain`] (no special-casing) | -//! | `""`, `ssl://`, `ssl://host`, `host:`, `:50001`, `host:abc`, `host:99999` | [`ParseServerAddrError`] | -//! | `host:0` | valid (port 0 is allowed) | -//! -//! [`ServerAddr`] parses the same `host:port` grammar, but rejects any input containing -//! `"://"`. -//! -//! Additional rules: -//! -//! * No trimming; no default ports. -//! * Domains are not validated beyond being non-empty; unresolvable names fail later at -//! connection time. -//! * The error payload is the full input string that failed to parse. -//! -//! # Connecting -//! -//! The actual transport constructors live in submodules: -//! -//! * [`blocking`]: blocking (std I/O) constructors, e.g. [`blocking::connect_tcp`]. -//! * [`tokio`]: Tokio-based async constructors, e.g. [`tokio::connect_tcp`] (feature `tokio`). - -use std::fmt; -use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; -use std::str::FromStr; - -/// The connection scheme of an Electrum server, taken from the URL prefix. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum Scheme { - /// Plaintext TCP (`tcp://`). - Tcp, - /// SSL/TLS encrypted TCP (`ssl://`). - Ssl, -} - -impl Scheme { - /// The URL prefix of this scheme (e.g. `tcp` for `tcp://`). - pub fn as_str(&self) -> &'static str { - match self { - Scheme::Tcp => "tcp", - Scheme::Ssl => "ssl", - } - } -} - -impl fmt::Display for Scheme { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.write_str(self.as_str()) - } -} - -impl FromStr for Scheme { - type Err = ParseServerAddrError; - - /// Parses a scheme prefix. Lowercase only: `"tcp"` and `"ssl"` are accepted. - fn from_str(s: &str) -> Result { - match s { - "tcp" => Ok(Scheme::Tcp), - "ssl" => Ok(Scheme::Ssl), - other => Err(ParseServerAddrError(other.to_string())), - } - } -} - -/// The host portion of a [`ServerAddr`]. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum Host { - /// A domain name, e.g. `electrum.example.com` or `….onion`. - 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"`). -/// Use [`ServerUrl`] to parse scheme-prefixed addresses. -#[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` or other SOCKS5 targets — those must be passed to the - /// proxy as an unresolved domain (proxy-side 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(), - } - } -} - -/// A full Electrum server URL: a connection [`Scheme`] and a [`ServerAddr`]. -/// -/// Parses from `"[scheme://]host:port"`. A missing scheme defaults to [`Scheme::Tcp`]. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ServerUrl { - scheme: Scheme, - addr: ServerAddr, -} - -impl ServerUrl { - /// Creates a new `ServerUrl` from a [`Scheme`] and [`ServerAddr`]. - pub fn new(scheme: Scheme, addr: ServerAddr) -> Self { - Self { scheme, addr } - } - - /// The connection scheme of this URL. - pub fn scheme(&self) -> Scheme { - self.scheme - } - - /// The server address (host and port) of this URL. - pub fn addr(&self) -> &ServerAddr { - &self.addr - } -} - -impl fmt::Display for ServerUrl { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}://{}", self.scheme, self.addr) - } -} - -impl FromStr for ServerUrl { - type Err = ParseServerAddrError; - - fn from_str(s: &str) -> Result { - let invalid = || ParseServerAddrError(s.to_string()); - let (scheme, addr_str) = match s.split_once("://") { - Some((scheme_str, addr_str)) => (scheme_str.parse().map_err(|_| invalid())?, addr_str), - None => (Scheme::Tcp, s), - }; - Ok(Self { - scheme, - addr: addr_str.parse().map_err(|_| invalid())?, - }) - } -} - -/// An error parsing a [`Scheme`], [`ServerAddr`] or [`ServerUrl`] 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 {} - -/// Credential for a SOCKS5 proxy. -#[derive(Clone, PartialEq, Eq)] -pub struct Socks5Credential { - /// Username for SOCKS5 authentication. - pub username: String, - /// Password for SOCKS5 authentication. - pub password: String, -} - -impl fmt::Debug for Socks5Credential { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("Socks5Credential") - .field("username", &self.username) - .field("password", &"") - .finish() - } -} - -/// Configuration of a SOCKS5 proxy, e.g. for connecting over Tor. -/// -/// The proxy address is typically a local address (e.g. `127.0.0.1:9050` for Tor). DNS -/// resolution of the *target* server happens proxy-side, which is what makes `.onion` -/// addresses reachable. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct Socks5Config { - /// The address of the SOCKS5 proxy. - pub addr: ServerAddr, - /// Credentials used to authenticate with the proxy, if any. - pub credentials: Option, -} - -impl Socks5Config { - /// Creates a new `Socks5Config` for a proxy that requires no authentication. - pub fn new(addr: ServerAddr) -> Self { - Self { - addr, - credentials: None, - } - } - - /// Creates a new `Socks5Config` for a proxy that requires username/password - /// authentication. - pub fn with_credentials(addr: ServerAddr, username: String, password: String) -> Self { - Self { - addr, - credentials: Some(Socks5Credential { username, password }), - } - } -} - -/// Configuration for establishing a connection to an Electrum server. -/// -/// 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, - /// The SOCKS5 proxy to connect through, if any. - socks5: Option, -} - -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 - } - - /// The SOCKS5 proxy to connect through, if any. - pub fn socks5(&self) -> Option<&Socks5Config> { - self.socks5.as_ref() - } -} - -impl Default for ConnectConfig { - fn default() -> Self { - Self { - timeout: None, - validate_domain: true, - socks5: None, - } - } -} - -/// 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 - } - - /// Sets the SOCKS5 proxy to connect through. See [`ConnectConfig::socks5`]. - pub fn socks5(mut self, socks5: Option) -> Self { - self.config.socks5 = socks5; - self - } - - /// Builds the [`ConnectConfig`]. - pub fn build(self) -> ConnectConfig { - self.config - } -} - -/// 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, - } - } -} diff --git a/src/transport/mod.rs b/src/transport/mod.rs new file mode 100644 index 0000000..6864dd3 --- /dev/null +++ b/src/transport/mod.rs @@ -0,0 +1,748 @@ +//! Address types and constructors for Electrum server connections. +//! +//! Hostnames stay unresolved ([`Host::Domain`]) for TLS SNI. +//! See [`ServerAddr`] / [`ServerUrl`] for parse rules. +//! +//! Constructors: [`blocking::connect_tcp`], [`tokio::connect_tcp`] (feature `tokio`). +//! TLS: [`blocking::connect_ssl`] (feature `ssl`) returns [`blocking::TlsStream`], which can be +//! split for full-duplex I/O; [`tokio::connect_ssl`] (features `ssl` and `tokio`). + +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 connection scheme of an Electrum server, taken from the URL prefix. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Scheme { + /// Plaintext TCP (`tcp://`). + Tcp, + /// SSL/TLS encrypted TCP (`ssl://`). + Ssl, +} + +impl Scheme { + /// The URL prefix of this scheme (e.g. `tcp` for `tcp://`). + pub fn as_str(&self) -> &'static str { + match self { + Scheme::Tcp => "tcp", + Scheme::Ssl => "ssl", + } + } +} + +impl fmt::Display for Scheme { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl FromStr for Scheme { + type Err = ParseServerAddrError; + + /// Parses a scheme prefix. Lowercase only: `"tcp"` and `"ssl"` are accepted. + fn from_str(s: &str) -> Result { + match s { + "tcp" => Ok(Scheme::Tcp), + "ssl" => Ok(Scheme::Ssl), + other => Err(ParseServerAddrError(other.to_string())), + } + } +} + +/// 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"`). +/// Use [`ServerUrl`] to parse scheme-prefixed addresses. +#[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(), + } + } +} + +/// A full Electrum server URL: a connection [`Scheme`] and a [`ServerAddr`]. +/// +/// Parses from `"[scheme://]host:port"`. A missing scheme defaults to [`Scheme::Tcp`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ServerUrl { + scheme: Scheme, + addr: ServerAddr, +} + +impl ServerUrl { + /// Creates a new `ServerUrl` from a [`Scheme`] and [`ServerAddr`]. + pub fn new(scheme: Scheme, addr: ServerAddr) -> Self { + Self { scheme, addr } + } + + /// The connection scheme of this URL. + pub fn scheme(&self) -> Scheme { + self.scheme + } + + /// The server address (host and port) of this URL. + pub fn addr(&self) -> &ServerAddr { + &self.addr + } +} + +impl fmt::Display for ServerUrl { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}://{}", self.scheme, self.addr) + } +} + +impl FromStr for ServerUrl { + type Err = ParseServerAddrError; + + fn from_str(s: &str) -> Result { + let invalid = || ParseServerAddrError(s.to_string()); + let (scheme, addr_str) = match s.split_once("://") { + Some((scheme_str, addr_str)) => (scheme_str.parse().map_err(|_| invalid())?, addr_str), + None => (Scheme::Tcp, s), + }; + Ok(Self { + scheme, + addr: addr_str.parse().map_err(|_| invalid())?, + }) + } +} + +/// An error parsing a [`Scheme`], [`ServerAddr`] or [`ServerUrl`] 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 {} + +/// Configuration for establishing a connection to an Electrum server. +/// +/// 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 a TLS connection. +#[cfg(feature = "ssl")] +#[non_exhaustive] +#[derive(Debug)] +pub enum TlsConnectError { + /// Transport or socket failure, including DNS, TCP, timeout, and handshake EOF/reset. + Io(std::io::Error), + /// TLS configuration, protocol, or certificate-validation failure. + Tls(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 TlsConnectError { + /// Extracts a wrapped `rustls::Error` so TLS failures are not misclassified as I/O. + fn from_handshake_io(err: std::io::Error) -> Self { + match err + .get_ref() + .and_then(|inner| inner.downcast_ref::()) + { + Some(tls) => TlsConnectError::Tls(tls.clone()), + None => TlsConnectError::Io(err), + } + } + + /// Server certificate rejected by the local verifier. + pub fn certificate_error(&self) -> Option<&rustls::CertificateError> { + match self { + Self::Tls(rustls::Error::InvalidCertificate(error)) => Some(error), + _ => None, + } + } +} + +#[cfg(feature = "ssl")] +impl fmt::Display for TlsConnectError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + TlsConnectError::Io(e) => write!(f, "connection I/O error: {e}"), + TlsConnectError::MissingDomain => { + write!(f, "TLS certificate validation requires a domain name") + } + TlsConnectError::InvalidServerName(name) => { + write!(f, "invalid TLS server name '{name}'") + } + TlsConnectError::Tls(e) => write!(f, "TLS error: {e}"), + } + } +} + +#[cfg(feature = "ssl")] +impl std::error::Error for TlsConnectError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + TlsConnectError::Io(e) => Some(e), + TlsConnectError::Tls(e) => Some(e), + TlsConnectError::MissingDomain | TlsConnectError::InvalidServerName(_) => None, + } + } +} + +#[cfg(feature = "ssl")] +impl From for TlsConnectError { + fn from(e: rustls::Error) -> Self { + TlsConnectError::Tls(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, TlsConnectError> { + match addr.host() { + Host::Domain(domain) => rustls::pki_types::ServerName::try_from(domain.clone()) + .map_err(|_| TlsConnectError::InvalidServerName(domain.clone())), + Host::Ip(ip) => { + if validate_domain { + Err(TlsConnectError::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; + + /// 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()), + } + } + + #[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::TlsConnectError::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::TlsConnectError::from_handshake_io)?; + + tcp.set_read_timeout(None) + .map_err(super::TlsConnectError::Io)?; + tcp.set_write_timeout(None) + .map_err(super::TlsConnectError::Io)?; + TlsStream::new(conn, tcp).map_err(super::TlsConnectError::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], + 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, + } + } + + /// 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::TlsConnectError> { + 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::TlsConnectError::Io)?; + connector + .connect(server_name, tcp) + .await + .map_err(super::TlsConnectError::from_handshake_io) + }; + match timeout { + Some(timeout) => match tokio::time::timeout(timeout, handshake).await { + Ok(res) => res, + Err(_elapsed) => Err(super::TlsConnectError::Io(io::Error::new( + io::ErrorKind::TimedOut, + format!("connection to '{}' timed out", addr), + ))), + }, + None => handshake.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}")) + } + + fn url(s: &str) -> ServerUrl { + 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}"); + } + } + + #[test] + fn server_url_parse() { + let u = url("127.0.0.1:50001"); + assert_eq!(u.scheme(), Scheme::Tcp); + assert_eq!(u.addr(), &addr("127.0.0.1:50001")); + assert_eq!(u.to_string(), "tcp://127.0.0.1:50001"); + + let u = url("tcp://electrum.example.com:50001"); + assert_eq!(u.scheme(), Scheme::Tcp); + assert_eq!( + u.addr().host(), + &Host::Domain("electrum.example.com".into()) + ); + + let u = url("ssl://[::1]:50002"); + assert_eq!(u.scheme(), Scheme::Ssl); + assert_eq!(u.addr().port(), 50002); + assert_eq!(u.to_string(), "ssl://[::1]:50002"); + + for bad in ["http://host:80", "TCP://host:1", "ssl://", "tcp://host"] { + assert!(bad.parse::().is_err(), "{bad}"); + } + } +} diff --git a/src/transport/tls.rs b/src/transport/tls.rs new file mode 100644 index 0000000..61d3a6b --- /dev/null +++ b/src/transport/tls.rs @@ -0,0 +1,311 @@ +//! Split-capable blocking rustls adapter for full-duplex I/O. +//! +//! `rustls::StreamOwned` has no independent read and write halves. Putting it +//! behind one mutex would serialize blocking reads and writes. This adapter +//! shares `ClientConnection` but releases its mutex before blocking on socket I/O. +//! +//! To avoid deadlocks, always lock TLS output before the rustls 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(()) +}