fedimint_connectors/
ws.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use fedimint_core::module::{ApiMethod, ApiRequestErased};
5#[cfg(not(target_family = "wasm"))]
6use fedimint_core::rustls::install_crypto_provider;
7use fedimint_core::util::{FmtCompact as _, SafeUrl};
8use fedimint_core::{apply, async_trait_maybe_send};
9use fedimint_logging::LOG_NET_WS;
10use jsonrpsee_core::client::ClientT;
11pub use jsonrpsee_core::client::Error as JsonRpcClientError;
12use jsonrpsee_types::ErrorCode;
13#[cfg(target_family = "wasm")]
14use jsonrpsee_wasm_client::{Client as WsClient, WasmClientBuilder as WsClientBuilder};
15#[allow(unused)]
16#[cfg(not(target_family = "wasm"))]
17use jsonrpsee_ws_client::{WsClient, WsClientBuilder};
18use serde_json::Value;
19use tracing::trace;
20pub type JsonRpcResult<T> = Result<T, JsonRpcClientError>;
21
22use super::Connector;
23use crate::error::ConnectorError;
24use crate::{
25    Connectivity, DynGatewayConnection, DynGuaridianConnection, IConnection, IGuardianConnection,
26    ServerError, ServerResult,
27};
28
29#[derive(Debug, Clone)]
30pub struct WebsocketConnector {}
31
32impl WebsocketConnector {
33    pub fn new() -> Self {
34        Self {}
35    }
36
37    async fn make_new_connection(
38        &self,
39        url: &SafeUrl,
40        api_secret: Option<&str>,
41    ) -> ServerResult<Arc<WsClient>> {
42        trace!(target: LOG_NET_WS, %url, "Creating new websocket connection");
43
44        #[cfg(not(target_family = "wasm"))]
45        let mut client = {
46            use jsonrpsee_ws_client::{CustomCertStore, WsClientBuilder};
47            use tokio_rustls::rustls::RootCertStore;
48
49            install_crypto_provider().await;
50            let webpki_roots = webpki_roots::TLS_SERVER_ROOTS.iter().cloned();
51            let mut root_certs = RootCertStore::empty();
52            root_certs.extend(webpki_roots);
53
54            let tls_cfg = CustomCertStore::builder()
55                .with_root_certificates(root_certs)
56                .with_no_client_auth();
57
58            WsClientBuilder::default()
59                .max_concurrent_requests(u16::MAX as usize)
60                .with_custom_cert_store(tls_cfg)
61        };
62
63        #[cfg(target_family = "wasm")]
64        let client = WsClientBuilder::default().max_concurrent_requests(u16::MAX as usize);
65
66        if let Some(api_secret) = api_secret {
67            #[cfg(not(target_family = "wasm"))]
68            {
69                // on native platforms, jsonrpsee-client ignores `user:pass@...` in the Url,
70                // but we can set up the headers manually
71
72                use base64::Engine as _;
73                use jsonrpsee_ws_client::{HeaderMap, HeaderValue};
74                let mut headers = HeaderMap::new();
75
76                let auth = base64::engine::general_purpose::STANDARD
77                    .encode(format!("fedimint:{api_secret}"));
78
79                headers.insert(
80                    "Authorization",
81                    HeaderValue::from_str(&format!("Basic {auth}")).expect("Can't fail"),
82                );
83
84                client = client.set_headers(headers);
85            }
86            #[cfg(target_family = "wasm")]
87            {
88                // on wasm, url will be handled by the browser, which should take care of
89                // `user:pass@...`
90                let mut url = url.clone();
91                url.set_username("fedimint")
92                    .map_err(|_| ServerError::InvalidEndpoint("Invalid username".into()))?;
93                url.set_password(Some(&api_secret))
94                    .map_err(|_| ServerError::InvalidEndpoint("Invalid secret".into()))?;
95
96                let client = client
97                    .build(url.as_str())
98                    .await
99                    .map_err(jsonrpc_error_to_peer_error)?;
100
101                return Ok(Arc::new(client));
102            }
103        }
104
105        let client = client
106            .build(url.as_str())
107            .await
108            .map_err(jsonrpc_error_to_peer_error)?;
109
110        Ok(Arc::new(client))
111    }
112}
113
114impl Default for WebsocketConnector {
115    fn default() -> Self {
116        Self::new()
117    }
118}
119
120#[async_trait::async_trait]
121impl Connector for WebsocketConnector {
122    async fn connect_guardian(
123        &self,
124        url: &SafeUrl,
125        api_secret: Option<&str>,
126    ) -> ServerResult<DynGuaridianConnection> {
127        let client = self.make_new_connection(url, api_secret).await?;
128        Ok(client.into_dyn())
129    }
130
131    async fn connect_gateway(
132        &self,
133        _url: &SafeUrl,
134    ) -> Result<DynGatewayConnection, ConnectorError> {
135        Err(ConnectorError::GatewayUnsupported)
136    }
137
138    fn connectivity(&self, _url: &SafeUrl) -> Connectivity {
139        Connectivity::Direct
140    }
141}
142
143#[apply(async_trait_maybe_send!)]
144impl IConnection for WsClient {
145    async fn await_disconnection(&self) {
146        self.on_disconnect().await;
147    }
148
149    fn is_connected(&self) -> bool {
150        WsClient::is_connected(self)
151    }
152}
153
154#[async_trait]
155impl IGuardianConnection for WsClient {
156    async fn request(&self, method: ApiMethod, request: ApiRequestErased) -> ServerResult<Value> {
157        let method = match method {
158            ApiMethod::Core(method) => method,
159            ApiMethod::Module(module_id, method) => format!("module_{module_id}_{method}"),
160        };
161
162        Ok(ClientT::request(self, &method, [request.to_json()])
163            .await
164            .map_err(jsonrpc_error_to_peer_error)?)
165    }
166}
167
168#[apply(async_trait_maybe_send!)]
169impl IConnection for Arc<WsClient> {
170    async fn await_disconnection(&self) {
171        self.on_disconnect().await;
172    }
173
174    fn is_connected(&self) -> bool {
175        WsClient::is_connected(self)
176    }
177}
178
179#[async_trait]
180impl IGuardianConnection for Arc<WsClient> {
181    async fn request(&self, method: ApiMethod, request: ApiRequestErased) -> ServerResult<Value> {
182        let method = match method {
183            ApiMethod::Core(method) => method,
184            ApiMethod::Module(module_id, method) => format!("module_{module_id}_{method}"),
185        };
186
187        Ok(
188            ClientT::request(self.as_ref(), &method, [request.to_json()])
189                .await
190                .map_err(jsonrpc_error_to_peer_error)?,
191        )
192    }
193}
194
195fn jsonrpc_error_to_peer_error(jsonrpc_error: JsonRpcClientError) -> ServerError {
196    match jsonrpc_error {
197        JsonRpcClientError::Call(error_object) => {
198            let message = error_object.message().to_owned();
199            match ErrorCode::from(error_object.code()) {
200                ErrorCode::ParseError | ErrorCode::OversizedRequest | ErrorCode::InvalidRequest => {
201                    ServerError::InvalidRequest(message)
202                }
203                ErrorCode::MethodNotFound => ServerError::InvalidRpcId(message),
204                ErrorCode::InvalidParams => ServerError::InvalidRequest(message),
205                ErrorCode::InternalError | ErrorCode::ServerIsBusy | ErrorCode::ServerError(_) => {
206                    ServerError::ServerError(message)
207                }
208            }
209        }
210        JsonRpcClientError::Transport(error) => ServerError::Transport(error),
211        JsonRpcClientError::RestartNeeded(arc) => {
212            ServerError::Transport(arc.fmt_compact().to_string().into())
213        }
214        JsonRpcClientError::ParseError(error) => {
215            ServerError::InvalidResponse(error.fmt_compact().to_string())
216        }
217        JsonRpcClientError::InvalidSubscriptionId => {
218            ServerError::Transport("Invalid subscription id".into())
219        }
220        JsonRpcClientError::InvalidRequestId(invalid_request_id) => {
221            ServerError::InvalidRequest(invalid_request_id.fmt_compact().to_string())
222        }
223        JsonRpcClientError::RequestTimeout => ServerError::Transport("Request timeout".into()),
224        JsonRpcClientError::Custom(e) => ServerError::Transport(e.into()),
225        JsonRpcClientError::HttpNotImplemented => {
226            ServerError::ServerError("Http not implemented".to_string())
227        }
228        JsonRpcClientError::EmptyBatchRequest(empty_batch_request) => {
229            ServerError::InvalidRequest(empty_batch_request.fmt_compact().to_string())
230        }
231        JsonRpcClientError::RegisterMethod(register_method_error) => {
232            ServerError::InvalidResponse(register_method_error.fmt_compact().to_string())
233        }
234    }
235}
fedimint_connectors/ws.rs

fedimint_connectors/
ws.rs
