foundry_common/provider/
runtime_transport.rs

1//! Runtime transport that connects on first request, which can take either of an HTTP,
2//! WebSocket, or IPC transport. Retries are handled by a client layer (e.g.,
3//! `RetryBackoffLayer`) when used.
4
5use crate::{
6    DEFAULT_USER_AGENT, REQUEST_TIMEOUT,
7    provider::{
8        mpp::transport::{LazyMppHttpTransport, lazy_mpp_ws_connect},
9        redact_url,
10    },
11};
12use alloy_json_rpc::{RequestPacket, ResponsePacket};
13use alloy_pubsub::{PubSubConnect, PubSubFrontend};
14use alloy_rpc_types_engine::{Claims, JwtSecret};
15use alloy_transport::{
16    Authorization, BoxTransport, TransportError, TransportErrorKind, TransportFut,
17    utils::guess_local_url,
18};
19use alloy_transport_ipc::IpcConnect;
20use alloy_transport_ws::WsConnect;
21use regex::{Captures, Regex};
22use reqwest::header::{HeaderName, HeaderValue};
23use std::{
24    error::Error as StdError,
25    fmt,
26    path::PathBuf,
27    str::FromStr,
28    sync::{Arc, LazyLock},
29};
30use thiserror::Error;
31use tokio::sync::RwLock;
32use tower::Service;
33use url::Url;
34
35/// Known MPP-enabled RPC host suffixes.
36///
37/// Endpoints matching these patterns always use the MPP WebSocket transport,
38/// regardless of whether local MPP keys have been discovered.
39const KNOWN_MPP_HOSTS: &[&str] = &[".mpp.tempo.xyz", ".mpp.moderato.tempo.xyz"];
40
41static HTTP_URL_RE: LazyLock<Regex> =
42    LazyLock::new(|| Regex::new(r#"(?i)https?://[^\s<>"']+"#).expect("valid URL regex"));
43
44/// An enum representing the different transports that can be used to connect to a runtime.
45/// Only meant to be used internally by [RuntimeTransport].
46#[derive(Clone, Debug)]
47pub enum InnerTransport {
48    /// HTTP transport with lazy MPP 402 handling.
49    ///
50    /// For known Tempo endpoints, the MPP layer additionally runs the
51    /// `wallet.tempo.xyz` device-code flow on a 402 when no local access key
52    /// is configured (see [`crate::tempo::ensure_access_key`]).
53    Http(LazyMppHttpTransport),
54    /// WebSocket transport
55    Ws(PubSubFrontend),
56    /// IPC transport
57    Ipc(PubSubFrontend),
58}
59
60/// Error type for the runtime transport.
61#[derive(Error, Debug)]
62pub enum RuntimeTransportError {
63    /// Internal transport error
64    #[error("Internal transport error: {0} with {1}")]
65    TransportError(TransportError, String),
66
67    /// Invalid URL scheme
68    #[error("URL scheme is not supported: {0}")]
69    BadScheme(String),
70
71    /// Invalid HTTP header
72    #[error("Invalid HTTP header: {0}")]
73    BadHeader(String),
74
75    /// Invalid file path
76    #[error("Invalid IPC file path: {0}")]
77    BadPath(String),
78
79    /// Invalid construction of Http provider
80    #[error(transparent)]
81    HttpConstructionError(#[from] reqwest::Error),
82
83    /// Invalid JWT
84    #[error("Invalid JWT: {0}")]
85    InvalidJwt(String),
86}
87
88/// Runtime transport that only connects on first request.
89///
90/// A runtime transport is a custom [`alloy_transport::Transport`] that only connects when the
91/// *first* request is made. When the first request is made, it will connect to the runtime using
92/// either an HTTP WebSocket, or IPC transport depending on the URL used.
93/// Retries for rate-limiting and timeout-related errors are handled by an external
94/// client layer (e.g., `RetryBackoffLayer`) when configured.
95#[derive(Clone, Debug)]
96pub struct RuntimeTransport {
97    /// The inner actual transport used.
98    inner: Arc<RwLock<Option<InnerTransport>>>,
99    /// The URL to connect to.
100    url: Url,
101    /// The headers to use for requests.
102    headers: Vec<String>,
103    /// The JWT to use for requests.
104    jwt: Option<String>,
105    /// The timeout for requests.
106    timeout: std::time::Duration,
107    /// Whether to accept invalid certificates.
108    accept_invalid_certs: bool,
109    /// Whether to disable automatic proxy detection.
110    no_proxy: bool,
111}
112
113/// A builder for [RuntimeTransport].
114#[derive(Debug)]
115pub struct RuntimeTransportBuilder {
116    url: Url,
117    headers: Vec<String>,
118    jwt: Option<String>,
119    timeout: std::time::Duration,
120    accept_invalid_certs: bool,
121    no_proxy: bool,
122}
123
124impl RuntimeTransportBuilder {
125    /// Create a new builder with the given URL.
126    pub const fn new(url: Url) -> Self {
127        Self {
128            url,
129            headers: vec![],
130            jwt: None,
131            timeout: REQUEST_TIMEOUT,
132            accept_invalid_certs: false,
133            no_proxy: false,
134        }
135    }
136
137    /// Set the URL for the transport.
138    pub fn with_headers(mut self, headers: Vec<String>) -> Self {
139        self.headers = headers;
140        self
141    }
142
143    /// Set the JWT for the transport.
144    pub fn with_jwt(mut self, jwt: Option<String>) -> Self {
145        self.jwt = jwt;
146        self
147    }
148
149    /// Set the timeout for the transport.
150    pub const fn with_timeout(mut self, timeout: std::time::Duration) -> Self {
151        self.timeout = timeout;
152        self
153    }
154
155    /// Set whether to accept invalid certificates.
156    pub const fn accept_invalid_certs(mut self, accept_invalid_certs: bool) -> Self {
157        self.accept_invalid_certs = accept_invalid_certs;
158        self
159    }
160
161    /// Set whether to disable automatic proxy detection.
162    ///
163    /// This can help in sandboxed environments (e.g., Cursor IDE sandbox, macOS App Sandbox)
164    /// where system proxy detection via SCDynamicStore causes crashes.
165    pub const fn no_proxy(mut self, no_proxy: bool) -> Self {
166        self.no_proxy = no_proxy;
167        self
168    }
169
170    /// Builds the [RuntimeTransport] and returns it in a disconnected state.
171    /// The runtime transport will then connect when the first request happens.
172    pub fn build(self) -> RuntimeTransport {
173        RuntimeTransport {
174            inner: Arc::new(RwLock::new(None)),
175            url: self.url,
176            headers: self.headers,
177            jwt: self.jwt,
178            timeout: self.timeout,
179            accept_invalid_certs: self.accept_invalid_certs,
180            no_proxy: self.no_proxy,
181        }
182    }
183}
184
185impl fmt::Display for RuntimeTransport {
186    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
187        write!(f, "RuntimeTransport {}", redact_url(self.url.as_str()))
188    }
189}
190
191impl RuntimeTransport {
192    /// Connects the underlying transport, depending on the URL scheme.
193    pub async fn connect(&self) -> Result<InnerTransport, RuntimeTransportError> {
194        match self.url.scheme() {
195            "http" | "https" => self.connect_http(),
196            "ws" | "wss" => self.connect_ws().await,
197            "file" => self.connect_ipc().await,
198            _ => Err(RuntimeTransportError::BadScheme(self.url.scheme().to_string())),
199        }
200    }
201
202    fn reqwest_headers(&self) -> Result<reqwest::header::HeaderMap, RuntimeTransportError> {
203        let mut headers = reqwest::header::HeaderMap::new();
204
205        // If there's a JWT, add it to the headers if we can decode it.
206        if let Some(jwt) = self.jwt.clone() {
207            let auth =
208                build_auth(jwt).map_err(|e| RuntimeTransportError::InvalidJwt(e.to_string()))?;
209
210            let mut auth_value: HeaderValue =
211                HeaderValue::from_str(&auth.to_string()).expect("Header should be valid string");
212            auth_value.set_sensitive(true);
213
214            headers.insert(reqwest::header::AUTHORIZATION, auth_value);
215        };
216
217        // Add any custom headers.
218        for header in &self.headers {
219            let make_err = || RuntimeTransportError::BadHeader(header.clone());
220
221            let (key, val) = header.split_once(':').ok_or_else(make_err)?;
222
223            headers.insert(
224                HeaderName::from_str(key.trim()).map_err(|_| make_err())?,
225                HeaderValue::from_str(val.trim()).map_err(|_| make_err())?,
226            );
227        }
228
229        if !headers.contains_key(reqwest::header::USER_AGENT) {
230            headers.insert(
231                reqwest::header::USER_AGENT,
232                HeaderValue::from_str(DEFAULT_USER_AGENT)
233                    .expect("User-Agent should be valid string"),
234            );
235        }
236
237        // If MPP_API_KEY is set, attach it as x-api-key for gated MPP proxies.
238        // Does not override an explicit x-api-key header from the user.
239        if !headers.contains_key(HeaderName::from_static("x-api-key"))
240            && let Ok(api_key) = std::env::var("MPP_API_KEY")
241        {
242            let api_key = api_key.trim();
243            if !api_key.is_empty() {
244                let mut value = HeaderValue::from_str(api_key)
245                    .map_err(|_| RuntimeTransportError::BadHeader("MPP_API_KEY".to_string()))?;
246                value.set_sensitive(true);
247                headers.insert(HeaderName::from_static("x-api-key"), value);
248            }
249        }
250
251        Ok(headers)
252    }
253
254    fn reqwest_client_with_headers(
255        &self,
256        headers: reqwest::header::HeaderMap,
257    ) -> Result<reqwest::Client, RuntimeTransportError> {
258        let mut client_builder = reqwest::Client::builder()
259            .timeout(self.timeout)
260            .danger_accept_invalid_certs(self.accept_invalid_certs);
261
262        // Disable automatic proxy detection if requested. This helps in sandboxed environments
263        // (e.g., Cursor IDE sandbox, macOS App Sandbox) where system proxy detection via
264        // SCDynamicStore causes crashes. See: https://github.com/foundry-rs/foundry/issues/12733
265        if self.no_proxy || guess_local_url(self.url.as_str()) {
266            client_builder = client_builder.no_proxy();
267        }
268
269        client_builder = client_builder.default_headers(headers);
270
271        Ok(client_builder.build()?)
272    }
273
274    /// Creates a new reqwest client from this transport.
275    pub fn reqwest_client(&self) -> Result<reqwest::Client, RuntimeTransportError> {
276        self.reqwest_client_with_headers(self.reqwest_headers()?)
277    }
278
279    /// Connects to an HTTP transport with lazy MPP 402 handling.
280    fn connect_http(&self) -> Result<InnerTransport, RuntimeTransportError> {
281        let headers = self.reqwest_headers()?;
282        let client = self.reqwest_client_with_headers(headers.clone())?;
283        Ok(InnerTransport::Http(LazyMppHttpTransport::lazy(client, self.url.clone(), headers)))
284    }
285
286    /// Connects to a WS transport.
287    ///
288    /// Uses the canonical Alloy MPP WebSocket transport when the endpoint is a
289    /// known MPP service.
290    /// Otherwise falls back to alloy's plain [`WsConnect`] with zero overhead.
291    async fn connect_ws(&self) -> Result<InnerTransport, RuntimeTransportError> {
292        let auth = self.jwt.as_ref().and_then(|jwt| build_auth(jwt.clone()).ok());
293
294        let service = if is_known_mpp_endpoint(&self.url) {
295            let mut ws = lazy_mpp_ws_connect(&self.url);
296            if let Some(auth) = auth {
297                ws = ws.with_auth(auth);
298            }
299            ws.into_service().await.map_err(|e| {
300                RuntimeTransportError::TransportError(e, redact_url(self.url.as_str()))
301            })?
302        } else {
303            let mut ws = WsConnect::new(self.url.to_string());
304            if let Some(auth) = auth {
305                ws = ws.with_auth(auth);
306            }
307            ws.into_service().await.map_err(|e| {
308                RuntimeTransportError::TransportError(e, redact_url(self.url.as_str()))
309            })?
310        };
311
312        Ok(InnerTransport::Ws(service))
313    }
314
315    /// Connects to an IPC transport.
316    async fn connect_ipc(&self) -> Result<InnerTransport, RuntimeTransportError> {
317        let path = url_to_file_path(&self.url)
318            .map_err(|_| RuntimeTransportError::BadPath(self.url.to_string()))?;
319        let ipc_connector = IpcConnect::new(path.clone());
320        let ipc = ipc_connector.into_service().await.map_err(|e| {
321            RuntimeTransportError::TransportError(e, path.clone().display().to_string())
322        })?;
323        Ok(InnerTransport::Ipc(ipc))
324    }
325
326    /// Sends a request using the underlying transport.
327    /// If this is the first request, it will connect to the appropriate transport depending on the
328    /// URL scheme. Retries are performed by an external client layer (e.g., `RetryBackoffLayer`),
329    /// if such a layer is configured by the caller.
330    /// For sending the actual request, this action is delegated down to the
331    /// underlying transport through Tower's [tower::Service::call]. See tower's [tower::Service]
332    /// trait for more information.
333    pub fn request(&self, req: RequestPacket) -> TransportFut<'static> {
334        let this = self.clone();
335        Box::pin(async move {
336            let mut inner = this.inner.read().await;
337            if inner.is_none() {
338                drop(inner);
339                {
340                    let mut inner_mut = this.inner.write().await;
341                    if inner_mut.is_none() {
342                        *inner_mut =
343                            Some(this.connect().await.map_err(TransportErrorKind::custom)?);
344                    }
345                }
346                inner = this.inner.read().await;
347            }
348
349            // SAFETY: We just checked that the inner transport exists.
350            match inner.clone().expect("must've been initialized") {
351                InnerTransport::Http(mut http) => http
352                    .call(req)
353                    .await
354                    .map_err(|error| redact_http_transport_error(error, &this.url)),
355                InnerTransport::Ws(mut ws) => ws.call(req).await,
356                InnerTransport::Ipc(mut ipc) => ipc.call(req).await,
357            }
358        })
359    }
360
361    /// Convert this transport into a boxed trait object.
362    pub fn boxed(self) -> BoxTransport
363    where
364        Self: Sized + Clone + Send + Sync + 'static,
365    {
366        BoxTransport::new(self)
367    }
368}
369
370/// Returns `true` if `url` points to a known MPP-enabled RPC service.
371fn is_known_mpp_endpoint(url: &Url) -> bool {
372    url.host_str().is_some_and(|host| KNOWN_MPP_HOSTS.iter().any(|suffix| host.ends_with(suffix)))
373}
374
375fn redact_http_transport_error(error: TransportError, endpoint: &Url) -> TransportError {
376    let alloy_json_rpc::RpcError::Transport(TransportErrorKind::Custom(source)) = &error else {
377        return error;
378    };
379    let safe_endpoint = redact_url(endpoint.as_str());
380
381    let mut message = String::new();
382    let mut error: Option<&(dyn StdError + 'static)> = Some(source.as_ref());
383    while let Some(source) = error {
384        if !message.is_empty() {
385            message.push_str(": ");
386        }
387        message.push_str(&source.to_string());
388        error = source.source();
389    }
390    let message = HTTP_URL_RE.replace_all(&message, |captures: &Captures<'_>| {
391        let candidate = &captures[0];
392        let Ok(url) = Url::parse(candidate) else { return candidate.to_owned() };
393        if url.host() == endpoint.host()
394            && url.port_or_known_default() == endpoint.port_or_known_default()
395        {
396            safe_endpoint.clone()
397        } else {
398            candidate.to_owned()
399        }
400    });
401    TransportErrorKind::custom_str(&message)
402}
403
404impl tower::Service<RequestPacket> for RuntimeTransport {
405    type Response = ResponsePacket;
406    type Error = TransportError;
407    type Future = TransportFut<'static>;
408
409    #[inline]
410    fn poll_ready(
411        &mut self,
412        _cx: &mut std::task::Context<'_>,
413    ) -> std::task::Poll<Result<(), Self::Error>> {
414        std::task::Poll::Ready(Ok(()))
415    }
416
417    #[inline]
418    fn call(&mut self, req: RequestPacket) -> Self::Future {
419        self.request(req)
420    }
421}
422
423impl tower::Service<RequestPacket> for &RuntimeTransport {
424    type Response = ResponsePacket;
425    type Error = TransportError;
426    type Future = TransportFut<'static>;
427
428    #[inline]
429    fn poll_ready(
430        &mut self,
431        _cx: &mut std::task::Context<'_>,
432    ) -> std::task::Poll<Result<(), Self::Error>> {
433        std::task::Poll::Ready(Ok(()))
434    }
435
436    #[inline]
437    fn call(&mut self, req: RequestPacket) -> Self::Future {
438        self.request(req)
439    }
440}
441
442fn build_auth(jwt: String) -> eyre::Result<Authorization> {
443    // Decode jwt from hex, then generate claims (iat with current timestamp)
444    let secret = JwtSecret::from_hex(jwt)?;
445    let claims = Claims::default();
446    let token = secret.encode(&claims)?;
447
448    let auth = Authorization::Bearer(token);
449
450    Ok(auth)
451}
452
453#[cfg(windows)]
454fn url_to_file_path(url: &Url) -> Result<PathBuf, ()> {
455    const PREFIX: &str = "file:///pipe/";
456
457    let url_str = url.as_str();
458
459    if let Some(pipe_name) = url_str.strip_prefix(PREFIX) {
460        let pipe_path = format!(r"\\.\pipe\{pipe_name}");
461        return Ok(PathBuf::from(pipe_path));
462    }
463
464    url.to_file_path()
465}
466
467#[cfg(not(windows))]
468fn url_to_file_path(url: &Url) -> Result<PathBuf, ()> {
469    url.to_file_path()
470}
471
472#[cfg(test)]
473mod tests {
474    use super::*;
475    use reqwest::header::HeaderMap;
476    use std::io;
477
478    #[derive(Debug, Error)]
479    #[error("request to https://example.com/private-api-key failed")]
480    struct ProviderError {
481        #[source]
482        source: io::Error,
483    }
484
485    #[test]
486    fn http_transport_errors_preserve_provider_guidance() {
487        let endpoint =
488            Url::parse("https://user:password@example.com/private-api-key?token=secret").unwrap();
489        let error = TransportErrorKind::custom(ProviderError {
490            source: io::Error::other(
491                "Authorize an access key with:\n  cast tempo login --no-browser",
492            ),
493        });
494
495        let report = redact_http_transport_error(error, &endpoint).to_string();
496
497        assert!(report.contains("https://example.com/"));
498        assert!(report.contains("cast tempo login --no-browser"));
499        assert!(!report.contains("password"));
500        assert!(!report.contains("private-api-key"));
501        assert!(!report.contains("secret"));
502    }
503
504    #[test]
505    fn http_transport_errors_redact_endpoint_paths() {
506        let endpoint =
507            Url::parse("https://user:password@example.com/private-api-key?token=secret").unwrap();
508        let error = TransportErrorKind::custom_str(concat!(
509            "request to https://example.com/private-api-key failed: connection refused\n\n",
510            "Authorize an access key with:\n  cast tempo login"
511        ));
512
513        let error = redact_http_transport_error(error, &endpoint);
514        let report = error.to_string();
515
516        assert!(report.contains("https://example.com/"));
517        assert!(!report.contains("password"));
518        assert!(!report.contains("private-api-key"));
519        assert!(!report.contains("secret"));
520        assert!(report.to_lowercase().contains("connection refused"));
521        assert!(report.contains("cast tempo login"));
522    }
523
524    #[test]
525    fn http_transport_errors_redact_normalized_endpoint_variants() {
526        let endpoint =
527            Url::parse("https://user:password@example.com/private-api-key?token=secret").unwrap();
528        let error = TransportErrorKind::custom_str(
529            "request to https://USER:normalized@example.com:443/different%2Fpath?key=other failed",
530        );
531
532        let report = redact_http_transport_error(error, &endpoint).to_string();
533
534        assert!(report.contains("https://example.com/"));
535        assert!(!report.contains("normalized"));
536        assert!(!report.contains("different"));
537        assert!(!report.contains("other"));
538    }
539
540    #[tokio::test]
541    async fn websocket_error_redacts_url_credentials() {
542        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
543        let address = listener.local_addr().unwrap();
544        drop(listener);
545        let url = Url::parse(&format!(
546            "ws://user:password@{address}/private-api-key?token=secret#fragment"
547        ))
548        .unwrap();
549        let transport = RuntimeTransportBuilder::new(url).build();
550
551        let error = transport.connect_ws().await.unwrap_err().to_string();
552
553        assert!(error.contains(&format!("ws://{address}/")));
554        assert!(!error.contains("user"));
555        assert!(!error.contains("password"));
556        assert!(!error.contains("private-api-key"));
557        assert!(!error.contains("secret"));
558    }
559
560    #[tokio::test]
561    async fn test_user_agent_header() {
562        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
563        let url = Url::parse(&format!("http://{}", listener.local_addr().unwrap())).unwrap();
564
565        let http_handler = axum::routing::get(|actual_headers: HeaderMap| {
566            let user_agent = HeaderName::from_str("User-Agent").unwrap();
567            assert_eq!(actual_headers[user_agent], HeaderValue::from_str("test-agent").unwrap());
568
569            async { "" }
570        });
571
572        let server_task = tokio::spawn(async move {
573            axum::serve(listener, http_handler.into_make_service()).await.unwrap()
574        });
575
576        let transport = RuntimeTransportBuilder::new(url.clone())
577            .with_headers(vec!["User-Agent: test-agent".to_string()])
578            .build();
579        let inner = transport.connect_http().unwrap();
580
581        match inner {
582            InnerTransport::Http(http) => {
583                let _ = http.client().get(url).send().await.unwrap();
584
585                // assert inside http_handler
586            }
587            _ => unreachable!(),
588        }
589
590        server_task.abort();
591    }
592}
foundry_common/provider/runtime_transport.rs

foundry_common/provider/
runtime_transport.rs
