phoenix_channel_runtime_native/
lib.rs

1#![doc = include_str!("../README.md")]
2#![forbid(unsafe_code)]
3#![warn(missing_docs)]
4#![warn(rustdoc::broken_intra_doc_links)]
5
6#[cfg(not(target_arch = "wasm32"))]
7mod native {
8    use std::{
9        sync::{Arc, OnceLock},
10        time::{Duration, Instant},
11    };
12
13    use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
14    use futures::{SinkExt, StreamExt};
15    use phoenix_channel_client::{ConnectContext, Connector, Endpoint, ResolvedEndpoint, Timer};
16    use phoenix_channel_runtime::{
17        Transport, TransportClose, TransportCloseRequest, TransportError, TransportErrorKind,
18        TransportEvent, WireMessage,
19    };
20    use tokio::{
21        io::{AsyncReadExt, AsyncWriteExt},
22        net::TcpStream,
23    };
24    use tokio_tungstenite::{
25        Connector as TlsConnector, MaybeTlsStream, WebSocketStream, client_async_tls_with_config,
26        tungstenite::{
27            Message,
28            client::IntoClientRequest,
29            http::{HeaderName, HeaderValue, header::SEC_WEBSOCKET_PROTOCOL},
30            protocol::{CloseFrame, WebSocketConfig, frame::coding::CloseCode},
31        },
32    };
33    use url::Url;
34
35    /// Tokio Tungstenite transport implementing the runtime-neutral transport API.
36    pub struct NativeTransport {
37        inner: WebSocketStream<MaybeTlsStream<TcpStream>>,
38    }
39
40    /// Connector for native `ws` and `wss` WebSocket endpoints.
41    #[derive(Clone)]
42    pub struct NativeConnector {
43        endpoint: NativeEndpoint,
44        options: NativeTransportOptions,
45    }
46
47    /// HTTP CONNECT proxy URL and host bypass rules.
48    #[derive(Clone)]
49    pub struct ProxyConfig {
50        url: String,
51        bypass_hosts: Vec<String>,
52    }
53
54    impl std::fmt::Debug for ProxyConfig {
55        fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
56            formatter
57                .debug_struct("ProxyConfig")
58                .field("url", &"configured")
59                .field("bypass_hosts", &self.bypass_hosts)
60                .finish()
61        }
62    }
63
64    impl ProxyConfig {
65        /// Creates proxy configuration for an HTTP proxy URL.
66        pub fn new(url: impl Into<String>) -> Self {
67            Self {
68                url: url.into(),
69                bypass_hosts: Vec::new(),
70            }
71        }
72
73        /// Bypasses the proxy for a host and all of its subdomains.
74        pub fn bypass_host(mut self, host: impl Into<String>) -> Self {
75            self.bypass_hosts.push(host.into());
76            self
77        }
78
79        fn applies_to(&self, host: &str) -> bool {
80            !self.bypass_hosts.iter().any(|entry| {
81                let entry = entry.trim_start_matches('.');
82                host.eq_ignore_ascii_case(entry)
83                    || host
84                        .to_ascii_lowercase()
85                        .ends_with(&format!(".{}", entry.to_ascii_lowercase()))
86            })
87        }
88    }
89
90    /// TLS, proxy, header, and WebSocket limit settings for native transports.
91    #[derive(Clone)]
92    pub struct NativeTransportOptions {
93        headers: Vec<(String, String)>,
94        tls_config: Option<Arc<rustls::ClientConfig>>,
95        proxy: Option<ProxyConfig>,
96        max_message_size: Option<usize>,
97        max_frame_size: Option<usize>,
98        disable_nagle: bool,
99    }
100
101    impl std::fmt::Debug for NativeTransportOptions {
102        fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
103            let header_names = self
104                .headers
105                .iter()
106                .map(|(name, _)| name)
107                .collect::<Vec<_>>();
108            formatter
109                .debug_struct("NativeTransportOptions")
110                .field("header_names", &header_names)
111                .field(
112                    "tls_config",
113                    &self.tls_config.as_ref().map(|_| "configured"),
114                )
115                .field("proxy", &self.proxy)
116                .field("max_message_size", &self.max_message_size)
117                .field("max_frame_size", &self.max_frame_size)
118                .field("disable_nagle", &self.disable_nagle)
119                .finish()
120        }
121    }
122
123    impl Default for NativeTransportOptions {
124        fn default() -> Self {
125            Self {
126                headers: Vec::new(),
127                tls_config: None,
128                proxy: None,
129                max_message_size: Some(16 * 1024 * 1024),
130                max_frame_size: Some(16 * 1024 * 1024),
131                disable_nagle: false,
132            }
133        }
134    }
135
136    impl NativeTransportOptions {
137        /// Adds an HTTP header to the WebSocket upgrade request.
138        pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
139            self.headers.push((name.into(), value.into()));
140            self
141        }
142
143        /// Uses an application-provided Rustls client configuration.
144        pub fn tls_config(mut self, config: Arc<rustls::ClientConfig>) -> Self {
145            self.tls_config = Some(config);
146            self
147        }
148
149        /// Routes applicable connections through an HTTP CONNECT proxy.
150        pub fn proxy(mut self, proxy: ProxyConfig) -> Self {
151            self.proxy = Some(proxy);
152            self
153        }
154
155        /// Sets the maximum accepted WebSocket message size, or disables it.
156        pub fn max_message_size(mut self, value: Option<usize>) -> Self {
157            self.max_message_size = value;
158            self
159        }
160
161        /// Sets the maximum accepted WebSocket frame size, or disables it.
162        pub fn max_frame_size(mut self, value: Option<usize>) -> Self {
163            self.max_frame_size = value;
164            self
165        }
166
167        /// Enables or disables TCP `nodelay` on direct and proxied connections.
168        pub fn disable_nagle(mut self, value: bool) -> Self {
169            self.disable_nagle = value;
170            self
171        }
172    }
173
174    #[derive(Clone)]
175    enum NativeEndpoint {
176        Url(String),
177        Phoenix(Endpoint),
178    }
179
180    impl NativeConnector {
181        /// Creates a connector for an already-resolved WebSocket URL.
182        pub fn new(url: impl Into<String>) -> Self {
183            Self {
184                endpoint: NativeEndpoint::Url(url.into()),
185                options: NativeTransportOptions::default(),
186            }
187        }
188
189        /// Creates a connector that resolves a Phoenix [`Endpoint`] each attempt.
190        pub fn from_endpoint(endpoint: Endpoint) -> Self {
191            Self {
192                endpoint: NativeEndpoint::Phoenix(endpoint),
193                options: NativeTransportOptions::default(),
194            }
195        }
196
197        /// Adds an HTTP header to the WebSocket upgrade request.
198        pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
199            self.options = self.options.header(name, value);
200            self
201        }
202
203        /// Replaces the native transport options.
204        pub fn options(mut self, options: NativeTransportOptions) -> Self {
205            self.options = options;
206            self
207        }
208    }
209
210    impl Connector for NativeConnector {
211        fn connect(
212            &self,
213            context: ConnectContext,
214        ) -> futures::future::LocalBoxFuture<'static, Result<Box<dyn Transport>, TransportError>>
215        {
216            let endpoint = self.endpoint.clone();
217            let options = self.options.clone();
218            Box::pin(async move {
219                let endpoint = match endpoint {
220                    NativeEndpoint::Url(url) => ResolvedEndpoint {
221                        url,
222                        protocols: Vec::new(),
223                    },
224                    NativeEndpoint::Phoenix(endpoint) => {
225                        endpoint.resolve(context).await.map_err(|error| {
226                            TransportError::with_kind(
227                                TransportErrorKind::Connect,
228                                error.to_string(),
229                            )
230                        })?
231                    }
232                };
233                let transport = NativeTransport::connect_resolved(endpoint, options).await?;
234                Ok(Box::new(transport) as Box<dyn Transport>)
235            })
236        }
237    }
238
239    /// Tokio timer using monotonic [`std::time::Instant`] measurements.
240    #[derive(Clone, Copy, Debug, Default)]
241    pub struct NativeTimer;
242
243    impl Timer for NativeTimer {
244        fn sleep(&self, duration: Duration) -> futures::future::LocalBoxFuture<'static, ()> {
245            Box::pin(tokio::time::sleep(duration))
246        }
247
248        fn now(&self) -> Duration {
249            static ORIGIN: OnceLock<Instant> = OnceLock::new();
250            ORIGIN.get_or_init(Instant::now).elapsed()
251        }
252    }
253
254    impl NativeTransport {
255        /// Opens an already-resolved WebSocket URL with default options.
256        pub async fn connect(url: &str) -> Result<Self, TransportError> {
257            Self::connect_resolved(
258                ResolvedEndpoint {
259                    url: url.to_owned(),
260                    protocols: Vec::new(),
261                },
262                NativeTransportOptions::default(),
263            )
264            .await
265        }
266
267        async fn connect_resolved(
268            endpoint: ResolvedEndpoint,
269            options: NativeTransportOptions,
270        ) -> Result<Self, TransportError> {
271            let mut request = endpoint.url.into_client_request().map_err(|error| {
272                TransportError::with_kind(TransportErrorKind::Connect, error.to_string())
273            })?;
274            for (name, value) in &options.headers {
275                let name = HeaderName::try_from(name).map_err(|error| {
276                    TransportError::with_kind(TransportErrorKind::Connect, error.to_string())
277                })?;
278                let value = HeaderValue::try_from(value).map_err(|error| {
279                    TransportError::with_kind(TransportErrorKind::Connect, error.to_string())
280                })?;
281                request.headers_mut().insert(name, value);
282            }
283            if !endpoint.protocols.is_empty() {
284                let value =
285                    HeaderValue::try_from(endpoint.protocols.join(", ")).map_err(|error| {
286                        TransportError::with_kind(TransportErrorKind::Connect, error.to_string())
287                    })?;
288                request.headers_mut().insert(SEC_WEBSOCKET_PROTOCOL, value);
289            }
290            let host = request
291                .uri()
292                .host()
293                .ok_or_else(|| connect_error("WebSocket URL has no host"))?
294                .to_owned();
295            let port = request.uri().port_u16().unwrap_or_else(|| {
296                if request.uri().scheme_str() == Some("wss") {
297                    443
298                } else {
299                    80
300                }
301            });
302            let stream = if let Some(proxy) = options
303                .proxy
304                .as_ref()
305                .filter(|proxy| proxy.applies_to(&host))
306            {
307                connect_proxy(proxy, &host, port).await?
308            } else {
309                TcpStream::connect((host.as_str(), port))
310                    .await
311                    .map_err(|error| connect_error(error.to_string()))?
312            };
313            if options.disable_nagle {
314                stream
315                    .set_nodelay(true)
316                    .map_err(|error| connect_error(error.to_string()))?;
317            }
318            let websocket_config = WebSocketConfig::default()
319                .max_message_size(options.max_message_size)
320                .max_frame_size(options.max_frame_size);
321            let tls = options.tls_config.map(TlsConnector::Rustls);
322            let (inner, _) =
323                client_async_tls_with_config(request, stream, Some(websocket_config), tls)
324                    .await
325                    .map_err(|error| {
326                        TransportError::with_kind(TransportErrorKind::Connect, error.to_string())
327                    })?;
328            Ok(Self { inner })
329        }
330    }
331
332    fn connect_error(message: impl Into<String>) -> TransportError {
333        TransportError::with_kind(TransportErrorKind::Connect, message)
334    }
335
336    async fn connect_proxy(
337        proxy: &ProxyConfig,
338        target_host: &str,
339        target_port: u16,
340    ) -> Result<TcpStream, TransportError> {
341        let url = Url::parse(&proxy.url).map_err(|error| connect_error(error.to_string()))?;
342        if url.scheme() != "http" {
343            return Err(connect_error("only HTTP CONNECT proxies are supported"));
344        }
345        let host = url
346            .host_str()
347            .ok_or_else(|| connect_error("proxy URL has no host"))?;
348        let port = url.port_or_known_default().unwrap_or(80);
349        let mut stream = TcpStream::connect((host, port))
350            .await
351            .map_err(|error| connect_error(error.to_string()))?;
352        let authority = if target_host.contains(':') {
353            format!("[{target_host}]:{target_port}")
354        } else {
355            format!("{target_host}:{target_port}")
356        };
357        let authorization = if url.username().is_empty() {
358            String::new()
359        } else {
360            let credentials = format!("{}:{}", url.username(), url.password().unwrap_or_default());
361            format!(
362                "Proxy-Authorization: Basic {}\r\n",
363                BASE64.encode(credentials)
364            )
365        };
366        let request =
367            format!("CONNECT {authority} HTTP/1.1\r\nHost: {authority}\r\n{authorization}\r\n");
368        stream
369            .write_all(request.as_bytes())
370            .await
371            .map_err(|error| connect_error(error.to_string()))?;
372        let mut response = Vec::with_capacity(1024);
373        let mut byte = [0_u8; 1];
374        while response.len() < 8192 && !response.ends_with(b"\r\n\r\n") {
375            let count = stream
376                .read(&mut byte)
377                .await
378                .map_err(|error| connect_error(error.to_string()))?;
379            if count == 0 {
380                return Err(connect_error("proxy closed before completing CONNECT"));
381            }
382            response.push(byte[0]);
383        }
384        if !response.ends_with(b"\r\n\r\n") {
385            return Err(connect_error(
386                "proxy CONNECT response headers are too large",
387            ));
388        }
389        let status = String::from_utf8_lossy(&response);
390        let accepted = status
391            .lines()
392            .next()
393            .and_then(|line| line.split_whitespace().nth(1))
394            .is_some_and(|code| code.starts_with('2'));
395        if !accepted {
396            return Err(connect_error(format!(
397                "proxy CONNECT failed: {}",
398                status.lines().next().unwrap_or("invalid response")
399            )));
400        }
401        Ok(stream)
402    }
403
404    impl Transport for NativeTransport {
405        fn send<'a>(
406            &'a mut self,
407            message: WireMessage,
408        ) -> futures::future::LocalBoxFuture<'a, Result<(), TransportError>> {
409            Box::pin(async move {
410                let message = match message {
411                    WireMessage::Text(text) => Message::Text(text.into()),
412                    WireMessage::Binary(bytes) => Message::Binary(bytes.into()),
413                };
414                self.inner.send(message).await.map_err(|error| {
415                    TransportError::with_kind(TransportErrorKind::Send, error.to_string())
416                })
417            })
418        }
419
420        fn receive<'a>(
421            &'a mut self,
422        ) -> futures::future::LocalBoxFuture<'a, Result<TransportEvent, TransportError>> {
423            Box::pin(async move {
424                loop {
425                    let Some(message) = self.inner.next().await else {
426                        return Ok(TransportEvent::Closed(TransportClose::connection_ended()));
427                    };
428                    let message = message.map_err(|error| {
429                        TransportError::with_kind(TransportErrorKind::Receive, error.to_string())
430                    })?;
431                    match message {
432                        Message::Text(text) => {
433                            return Ok(TransportEvent::Message(WireMessage::Text(
434                                text.to_string(),
435                            )));
436                        }
437                        Message::Binary(bytes) => {
438                            return Ok(TransportEvent::Message(WireMessage::Binary(
439                                bytes.to_vec(),
440                            )));
441                        }
442                        Message::Close(frame) => {
443                            return Ok(TransportEvent::Closed(match frame {
444                                Some(frame) => {
445                                    let code = u16::from(frame.code);
446                                    TransportClose::new(
447                                        Some(code),
448                                        frame.reason.to_string(),
449                                        code == 1000,
450                                    )
451                                }
452                                None => TransportClose::connection_ended(),
453                            }));
454                        }
455                        Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => continue,
456                    }
457                }
458            })
459        }
460
461        fn close<'a>(
462            &'a mut self,
463        ) -> futures::future::LocalBoxFuture<'a, Result<(), TransportError>> {
464            Box::pin(async move {
465                SinkExt::close(&mut self.inner).await.map_err(|error| {
466                    TransportError::with_kind(TransportErrorKind::Close, error.to_string())
467                })
468            })
469        }
470
471        fn close_with<'a>(
472            &'a mut self,
473            request: TransportCloseRequest,
474        ) -> futures::future::LocalBoxFuture<'a, Result<(), TransportError>> {
475            Box::pin(async move {
476                self.inner
477                    .send(Message::Close(Some(CloseFrame {
478                        code: CloseCode::from(request.code),
479                        reason: request.reason.into(),
480                    })))
481                    .await
482                    .map_err(|error| {
483                        TransportError::with_kind(TransportErrorKind::Close, error.to_string())
484                    })
485            })
486        }
487    }
488
489    #[cfg(test)]
490    mod tests {
491        use super::*;
492        use tokio::net::TcpListener;
493
494        #[test]
495        fn proxy_bypass_matches_hosts_and_subdomains() {
496            let proxy = ProxyConfig::new("http://localhost:8080").bypass_host("example.com");
497            assert!(!proxy.applies_to("example.com"));
498            assert!(!proxy.applies_to("api.example.com"));
499            assert!(proxy.applies_to("notexample.com"));
500        }
501
502        #[tokio::test(flavor = "current_thread")]
503        async fn establishes_authenticated_http_connect_tunnels() {
504            let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
505            let address = listener.local_addr().unwrap();
506            let server = tokio::spawn(async move {
507                let (mut stream, _) = listener.accept().await.unwrap();
508                let mut request = Vec::new();
509                let mut byte = [0_u8; 1];
510                while !request.ends_with(b"\r\n\r\n") {
511                    stream.read_exact(&mut byte).await.unwrap();
512                    request.push(byte[0]);
513                }
514                stream
515                    .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n")
516                    .await
517                    .unwrap();
518                String::from_utf8(request).unwrap()
519            });
520            let proxy = ProxyConfig::new(format!("http://user:pass@{address}"));
521            let _stream = connect_proxy(&proxy, "example.com", 443).await.unwrap();
522            let request = server.await.unwrap();
523            assert!(request.starts_with("CONNECT example.com:443 HTTP/1.1\r\n"));
524            assert!(request.contains("Proxy-Authorization: Basic dXNlcjpwYXNz\r\n"));
525        }
526    }
527}
528
529#[cfg(not(target_arch = "wasm32"))]
530mod managed;
531
532#[cfg(not(target_arch = "wasm32"))]
533pub use native::{
534    NativeConnector, NativeTimer, NativeTransport, NativeTransportOptions, ProxyConfig,
535};
536
537#[cfg(not(target_arch = "wasm32"))]
538pub use managed::*;
phoenix_channel_runtime_native/lib.rs

phoenix_channel_runtime_native/
lib.rs
