fix(wallet): route Tor websocket subscriptions through Arti

Author: thesimplekid

crates/cdk-http-client/src/transport/mod.rs

@@ -16,6 +16,15 @@ use crate::{HttpClient, HttpClientBuilder};
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 pub trait Transport: Default + Send + Sync + Debug + Clone {
+    /// Connect to a WebSocket endpoint using this transport.
+    async fn ws_connect(
+        &self,
+        url: &str,
+        headers: &[(&str, &str)],
+    ) -> Result<(crate::ws::WsSender, crate::ws::WsReceiver), crate::ws::WsError> {
+        crate::ws::connect(url, headers).await
+    }
+
     /// Make the transport use a proxy.
     ///
     /// SOCKS proxy schemes such as `socks5h` are available only when this crate

crates/cdk-http-client/src/transport/tor_transport.rs

@@ -295,6 +295,22 @@ impl Transport for TorAsync {
         Ok(txts)
     }
 
+    async fn ws_connect(
+        &self,
+        url: &str,
+        headers: &[(&str, &str)],
+    ) -> Result<(crate::ws::WsSender, crate::ws::WsReceiver), crate::ws::WsError> {
+        let parsed_url = Url::parse(url)
+            .map_err(|e| crate::ws::WsError::Connection(format!("Invalid URL: {e}")))?;
+        let pool = self
+            .ensure_pool()
+            .await
+            .map_err(|e| crate::ws::WsError::Connection(e.to_string()))?;
+        let idx = self.index_for_request(&http::Method::GET, &parsed_url, None, pool.len());
+
+        crate::ws::connect_tor(pool[idx].clone(), url, headers).await
+    }
+
     async fn http_get<R>(&self, url: Url, auth: Option<AuthToken>) -> Result<R, HttpError>
     where
         R: DeserializeOwned,

crates/cdk-http-client/src/ws/mod.rs

@@ -10,6 +10,8 @@ mod native;
 mod wasm;
 
 pub use error::WsError;
+#[cfg(all(feature = "tor", not(target_arch = "wasm32")))]
+pub(crate) use native::connect_tor;
 #[cfg(not(target_arch = "wasm32"))]
 pub use native::{connect, WsReceiver, WsSender};
 #[cfg(target_arch = "wasm32")]

crates/cdk-http-client/src/ws/native.rs

@@ -102,3 +102,55 @@ pub async fn connect(
         },
     ))
 }
+
+/// Connect to a WebSocket endpoint through an Arti Tor client.
+#[cfg(feature = "tor")]
+pub(crate) async fn connect_tor(
+    tor_client: arti_client::TorClient<tor_rtcompat::PreferredRuntime>,
+    url: &str,
+    headers: &[(&str, &str)],
+) -> Result<(WsSender, WsReceiver), WsError> {
+    let parsed_url =
+        url::Url::parse(url).map_err(|e| WsError::Connection(format!("Invalid URL: {e}")))?;
+
+    let host = parsed_url
+        .host_str()
+        .ok_or_else(|| WsError::Connection("WebSocket URL must include a host".to_string()))?;
+    let port = parsed_url
+        .port_or_known_default()
+        .ok_or_else(|| WsError::Connection("WebSocket URL must include a port".to_string()))?;
+
+    let mut request = url
+        .into_client_request()
+        .map_err(|e| WsError::Connection(e.to_string()))?;
+
+    for &(name, value) in headers {
+        if let (Ok(header_name), Ok(header_value)) = (
+            name.parse::<tokio_tungstenite::tungstenite::http::header::HeaderName>(),
+            value.parse::<tokio_tungstenite::tungstenite::http::header::HeaderValue>(),
+        ) {
+            request.headers_mut().insert(header_name, header_value);
+        }
+    }
+
+    let stream = tor_client
+        .connect((host, port))
+        .await
+        .map_err(|e| WsError::Connection(e.to_string()))?;
+
+    let (ws_stream, _) =
+        tokio_tungstenite::client_async_tls_with_config(request, stream, None, None)
+            .await
+            .map_err(|e| WsError::Connection(e.to_string()))?;
+
+    let (sink, stream) = ws_stream.split();
+
+    Ok((
+        WsSender {
+            inner: Box::new(sink),
+        },
+        WsReceiver {
+            inner: Box::new(stream),
+        },
+    ))
+}

crates/cdk/src/wallet/mint_connector/http_client.rs

@@ -311,6 +311,20 @@ impl<T> MintConnector for HttpClient<T>
 where
     T: Transport + Send + Sync + 'static,
 {
+    async fn connect_websocket(
+        &self,
+        url: &str,
+        headers: &[(&str, &str)],
+    ) -> Result<
+        (
+            cdk_common::ws_client::WsSender,
+            cdk_common::ws_client::WsReceiver,
+        ),
+        cdk_common::ws_client::WsError,
+    > {
+        self.transport.ws_connect(url, headers).await
+    }
+
     #[cfg(all(feature = "bip353", not(target_arch = "wasm32")))]
     #[instrument(skip(self), fields(mint_url = %self.mint_url))]
     async fn resolve_dns_txt(&self, domain: &str) -> Result<Vec<String>, Error> {

crates/cdk/src/wallet/mint_connector/mod.rs

@@ -36,6 +36,21 @@ pub type TorHttpClient = http_client::HttpClient<transport::TorAsync>;
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 pub trait MintConnector: Debug {
+    /// Connect to a WebSocket endpoint using the connector's transport.
+    async fn connect_websocket(
+        &self,
+        url: &str,
+        headers: &[(&str, &str)],
+    ) -> Result<
+        (
+            cdk_common::ws_client::WsSender,
+            cdk_common::ws_client::WsReceiver,
+        ),
+        cdk_common::ws_client::WsError,
+    > {
+        cdk_common::ws_client::connect(url, headers).await
+    }
+
     #[cfg(all(feature = "bip353", not(target_arch = "wasm32")))]
     /// Resolve the DNS record getting the TXT value
     async fn resolve_dns_txt(&self, _domain: &str) -> Result<Vec<String>, Error>;

crates/cdk/src/wallet/subscription.rs

@@ -21,7 +21,7 @@ use cdk_common::pub_sub::remote_consumer::{
 };
 use cdk_common::pub_sub::{Error as PubsubError, Spec, Subscriber};
 use cdk_common::subscription::WalletParams;
-use cdk_common::ws_client::{connect as ws_connect, WsError};
+use cdk_common::ws_client::WsError;
 use cdk_common::{CheckStateRequest, Method, PaymentMethod, RoutePath};
 use tokio::sync::mpsc;
 use uuid::Uuid;
@@ -528,10 +528,14 @@ async fn stream_client(
     let header_refs: Vec<(&str, &str)> = headers.iter().map(|(k, v)| (*k, v.as_str())).collect();
 
     tracing::debug!("Connecting to {}", url);
-    let (mut sender, mut receiver) = ws_connect(&url_str, &header_refs).await.map_err(|err| {
-        tracing::error!("Error connecting: {err:?}");
-        map_ws_error(err)
-    })?;
+    let (mut sender, mut receiver) = client
+        .http_client
+        .connect_websocket(&url_str, &header_refs)
+        .await
+        .map_err(|err| {
+            tracing::error!("Error connecting: {err:?}");
+            map_ws_error(err)
+        })?;
 
     tracing::debug!("Connected to {}", url);