nautilus_network/websocket/
proxy.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! HTTP `CONNECT` tunneling for outbound WebSocket connections.
17//!
18//! HTTP and HTTPS proxy URLs are supported. An HTTPS proxy adds TLS to the proxy hop; a `wss`
19//! target adds a separate TLS session after the tunnel is established. URL user information
20//! becomes Basic proxy authentication, and credential-bearing values are redacted from `Debug`
21//! output.
22//!
23//! The tunnel accepts only a `2xx` response and bounds response headers before parsing. It returns
24//! a stream positioned for the WebSocket handshake rather than performing that handshake itself.
25//!
26//! SOCKS URLs are recognized but not tunneled: the client logs a warning and connects directly.
27//! Both transport backends tunnel through the same [`ProxiedStream`], which implements the IO
28//! traits over every tunnel shape so each backend runs its own handshake over the finished stream.
29
30use std::{
31    fmt::Debug,
32    io,
33    pin::Pin,
34    task::{Context, Poll},
35};
36
37use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
38use nautilus_core::string::secret::REDACTED;
39use rustls::{ClientConfig, RootCertStore, pki_types::ServerName};
40use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
41use tokio_rustls::{TlsConnector, client::TlsStream};
42use url::Url;
43
44use crate::{net::TcpStream, transport::TransportError};
45
46/// Maximum size of a `CONNECT` proxy response we are willing to read.
47///
48/// Bounds the buffer so a malicious or broken proxy cannot make us allocate
49/// indefinitely while we wait for the header terminator.
50const MAX_PROXY_RESPONSE_BYTES: usize = 16 * 1024;
51
52/// Validated HTTP or HTTPS proxy URL.
53///
54/// The underlying URL is intentionally redacted from [`Debug`] output because
55/// URL user-info can contain proxy credentials.
56#[derive(Clone, PartialEq, Eq)]
57pub struct ProxyUrl(String);
58
59impl ProxyUrl {
60    /// Parses and validates an HTTP or HTTPS proxy URL.
61    ///
62    /// # Errors
63    ///
64    /// Returns an error if the URL is malformed, has no host, or uses an unsupported scheme.
65    pub fn parse(value: impl Into<String>) -> Result<Self, TransportError> {
66        let value = value.into();
67        ProxyTarget::parse(&value)?;
68        Ok(Self(value))
69    }
70
71    /// Returns the validated URL for transport configuration.
72    #[must_use]
73    pub fn expose(&self) -> &str {
74        &self.0
75    }
76}
77
78impl Debug for ProxyUrl {
79    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        f.debug_tuple(stringify!(ProxyUrl))
81            .field(&REDACTED)
82            .finish()
83    }
84}
85
86/// Stream produced by `tunnel_via_proxy` when the upstream is `ws://`
87/// (no upstream TLS, but the proxy hop itself may have been TLS-protected).
88///
89/// The TLS-bearing variants are boxed because [`tokio_rustls::client::TlsStream`]
90/// is large enough that a flat enum trips `clippy::large_enum_variant`. Boxing
91/// keeps the discriminant cheap to move while leaving the rare TLS path on the
92/// heap.
93#[derive(Debug)]
94pub enum ProxiedStream {
95    /// Plain TCP after a plain proxy hop.
96    Plain(TcpStream),
97    /// Plain TCP after a TLS proxy hop.
98    PlainOverTlsProxy(Box<TlsStream<TcpStream>>),
99    /// Upstream TLS over a plain proxy hop.
100    Tls(Box<TlsStream<TcpStream>>),
101    /// Upstream TLS over a TLS proxy hop.
102    TlsOverTlsProxy(Box<TlsStream<TlsStream<TcpStream>>>),
103}
104
105/// Combines the two IO traits so [`ProxiedStream`] resolves its variant in one place.
106trait ProxiedIo: AsyncRead + AsyncWrite + Send + Unpin {}
107
108impl<T: AsyncRead + AsyncWrite + Send + Unpin> ProxiedIo for T {}
109
110impl ProxiedStream {
111    fn inner_mut(&mut self) -> &mut dyn ProxiedIo {
112        match self {
113            Self::Plain(s) => s,
114            Self::PlainOverTlsProxy(s) | Self::Tls(s) => s.as_mut(),
115            Self::TlsOverTlsProxy(s) => s.as_mut(),
116        }
117    }
118}
119
120impl AsyncRead for ProxiedStream {
121    fn poll_read(
122        self: Pin<&mut Self>,
123        cx: &mut Context<'_>,
124        buf: &mut ReadBuf<'_>,
125    ) -> Poll<io::Result<()>> {
126        Pin::new(self.get_mut().inner_mut()).poll_read(cx, buf)
127    }
128}
129
130impl AsyncWrite for ProxiedStream {
131    fn poll_write(
132        self: Pin<&mut Self>,
133        cx: &mut Context<'_>,
134        buf: &[u8],
135    ) -> Poll<io::Result<usize>> {
136        Pin::new(self.get_mut().inner_mut()).poll_write(cx, buf)
137    }
138
139    fn poll_write_vectored(
140        self: Pin<&mut Self>,
141        cx: &mut Context<'_>,
142        bufs: &[io::IoSlice<'_>],
143    ) -> Poll<io::Result<usize>> {
144        Pin::new(self.get_mut().inner_mut()).poll_write_vectored(cx, bufs)
145    }
146
147    fn is_write_vectored(&self) -> bool {
148        match self {
149            Self::Plain(s) => s.is_write_vectored(),
150            Self::PlainOverTlsProxy(s) | Self::Tls(s) => s.is_write_vectored(),
151            Self::TlsOverTlsProxy(s) => s.is_write_vectored(),
152        }
153    }
154
155    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
156        Pin::new(self.get_mut().inner_mut()).poll_flush(cx)
157    }
158
159    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
160        Pin::new(self.get_mut().inner_mut()).poll_shutdown(cx)
161    }
162}
163
164/// Parsed components of a target WebSocket URL needed by the proxy hop.
165#[derive(Debug, Clone, PartialEq, Eq)]
166pub struct WsTarget {
167    /// Host name for DNS / SNI / `CONNECT` request line.
168    pub host: String,
169    /// TCP port of the WebSocket origin.
170    pub port: u16,
171    /// `true` when the WebSocket scheme is `wss://`.
172    pub is_tls: bool,
173}
174
175impl WsTarget {
176    /// Parses a `ws://` or `wss://` URL into the host, port, and TLS components.
177    ///
178    /// # Errors
179    ///
180    /// Returns an error if the URL fails to parse, is missing a hostname, or uses a scheme other
181    /// than `ws`/`wss`.
182    pub fn parse(url: &str) -> Result<Self, TransportError> {
183        let parsed = Url::parse(url)
184            .map_err(|e| TransportError::InvalidUrl(format!("invalid WebSocket URL: {e}")))?;
185
186        let is_tls = match parsed.scheme() {
187            "ws" => false,
188            "wss" => true,
189            other => {
190                return Err(TransportError::InvalidUrl(format!(
191                    "expected ws:// or wss:// scheme, was {other}"
192                )));
193            }
194        };
195
196        let raw_host = parsed
197            .host_str()
198            .ok_or_else(|| TransportError::InvalidUrl("missing hostname".to_string()))?;
199
200        // url::Url stores IPv6 literals in bracketed form (`[::1]`); the
201        // `CONNECT` request line and TLS SNI both want the unbracketed form.
202        let host = if raw_host.starts_with('[') && raw_host.ends_with(']') {
203            raw_host[1..raw_host.len() - 1].to_string()
204        } else {
205            raw_host.to_string()
206        };
207
208        let port = parsed.port().unwrap_or(if is_tls { 443 } else { 80 });
209
210        Ok(Self { host, port, is_tls })
211    }
212}
213
214/// Outcome of parsing a proxy URL prior to opening a tunnel.
215///
216/// SOCKS schemes are recognized but not implemented for the WebSocket path
217/// yet. They are surfaced as [`ProxyKind::Unsupported`] so callers can log
218/// a warning and fall back to a direct connection, preserving compatibility
219/// with REST configs that already pointed at a SOCKS proxy.
220#[derive(Debug, Clone, PartialEq, Eq)]
221pub enum ProxyKind {
222    /// HTTP / HTTPS forward proxy reachable via `CONNECT` tunneling.
223    Http(ProxyTarget),
224    /// Recognized scheme without a working tunnel (currently SOCKS).
225    Unsupported {
226        /// Original URL scheme (e.g. `socks5`).
227        scheme: String,
228    },
229}
230
231impl ProxyKind {
232    /// Parses a proxy URL into a [`ProxyKind`]. Returns
233    /// [`TransportError::InvalidUrl`] for malformed input or non-proxy
234    /// schemes (`ftp://`, `ws://`, etc.).
235    ///
236    /// # Errors
237    ///
238    /// Returns an error if [`ProxyTarget::parse`] fails.
239    pub fn parse(url: &str) -> Result<Self, TransportError> {
240        let parsed = Url::parse(url)
241            .map_err(|e| TransportError::InvalidUrl(format!("invalid proxy URL: {e}")))?;
242
243        match parsed.scheme() {
244            "http" | "https" => ProxyTarget::parse(url).map(ProxyKind::Http),
245            scheme @ ("socks5" | "socks5h" | "socks4" | "socks4a") => {
246                // Reject malformed inputs like `socks5:host:port` that parse as
247                // scheme + opaque path with no authority: surfacing them as
248                // Unsupported would silently fall back to a direct connection
249                // and hide the typo.
250                if parsed.host_str().is_none_or(str::is_empty) {
251                    return Err(TransportError::InvalidUrl(format!(
252                        "proxy URL is missing a host (did you mean {scheme}://...)?"
253                    )));
254                }
255                Ok(Self::Unsupported {
256                    scheme: scheme.to_string(),
257                })
258            }
259            other => Err(TransportError::InvalidUrl(format!(
260                "unsupported proxy scheme '{other}'; expected http:// or https://"
261            ))),
262        }
263    }
264}
265
266/// Parsed components of a forward proxy URL.
267#[derive(Clone, PartialEq, Eq)]
268pub struct ProxyTarget {
269    /// Host name of the proxy (used for both DNS and TLS SNI when
270    /// [`ProxyTarget::is_tls`] is `true`).
271    pub host: String,
272    /// TCP port of the proxy.
273    pub port: u16,
274    /// `true` when the proxy URL scheme is `https`.
275    pub is_tls: bool,
276    /// Pre-computed `Proxy-Authorization` header value, if the URL embeds
277    /// `user:pass@`.
278    pub auth_header: Option<String>,
279}
280
281impl Debug for ProxyTarget {
282    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
283        f.debug_struct(stringify!(ProxyTarget))
284            .field("host", &self.host)
285            .field("port", &self.port)
286            .field("is_tls", &self.is_tls)
287            .field("auth_header", &self.auth_header.as_ref().map(|_| REDACTED))
288            .finish()
289    }
290}
291
292impl ProxyTarget {
293    /// Parses a proxy URL into the components needed to establish the tunnel.
294    ///
295    /// Only `http://` and `https://` schemes are accepted here. Use
296    /// [`ProxyKind::parse`] when callers need to distinguish recognized but
297    /// unsupported schemes (currently SOCKS) from malformed input.
298    ///
299    /// # Errors
300    ///
301    /// Returns an error if the URL is malformed, the host is missing, or the scheme is neither
302    /// `http` nor `https`.
303    pub fn parse(url: &str) -> Result<Self, TransportError> {
304        let parsed = Url::parse(url)
305            .map_err(|e| TransportError::InvalidUrl(format!("invalid proxy URL: {e}")))?;
306
307        let is_tls = match parsed.scheme() {
308            "http" => false,
309            "https" => true,
310            "socks5" | "socks5h" | "socks4" | "socks4a" => {
311                return Err(TransportError::InvalidUrl(format!(
312                    "SOCKS proxy scheme '{}' is not yet supported for WebSocket connections; \
313                    use an http:// or https:// proxy",
314                    parsed.scheme()
315                )));
316            }
317            other => {
318                return Err(TransportError::InvalidUrl(format!(
319                    "unsupported proxy scheme '{other}'; expected http:// or https://"
320                )));
321            }
322        };
323
324        let raw_host = parsed
325            .host_str()
326            .ok_or_else(|| TransportError::InvalidUrl("proxy URL missing hostname".to_string()))?;
327
328        // url::Url stores IPv6 literals bracketed (`[::1]`); the bracketed
329        // form is only valid in the HTTP `Host:` header, not for DNS or
330        // TLS SNI, so we keep both representations.
331        let host = if raw_host.starts_with('[') && raw_host.ends_with(']') {
332            raw_host[1..raw_host.len() - 1].to_string()
333        } else {
334            raw_host.to_string()
335        };
336
337        let port = parsed.port().unwrap_or(if is_tls { 443 } else { 80 });
338
339        let auth_header = if parsed.username().is_empty() && parsed.password().is_none() {
340            None
341        } else {
342            let username = decode_userinfo(parsed.username());
343            let password = decode_userinfo(parsed.password().unwrap_or(""));
344            let credentials = format!("{username}:{password}");
345            Some(format!("Basic {}", BASE64.encode(credentials)))
346        };
347
348        Ok(Self {
349            host,
350            port,
351            is_tls,
352            auth_header,
353        })
354    }
355}
356
357/// Percent-decode a userinfo field from a proxy URL. `url::Url` keeps the
358/// raw percent-encoded form, so we decode it here before assembling the
359/// `Basic` credentials.
360fn decode_userinfo(value: &str) -> String {
361    let bytes = nautilus_core::string::urlencoding::decode_bytes(value.as_bytes());
362    String::from_utf8_lossy(&bytes).into_owned()
363}
364
365/// Establish a tunneled connection through `proxy` to the WebSocket `target`.
366///
367/// On success the returned stream is positioned right after the proxy's
368/// `200`/`2xx` response, ready for the WebSocket handshake. The function does
369/// not perform the WebSocket handshake itself; callers wrap the stream in
370/// `tokio-tungstenite::client_async`.
371///
372/// # Errors
373///
374/// Returns an error if:
375/// - The TCP connection to the proxy fails.
376/// - The TLS layer to the proxy or upstream cannot be established.
377/// - The proxy returns a non-success status, malformed headers, or closes the
378///   stream before completing the response.
379pub async fn tunnel_via_proxy(
380    target: &WsTarget,
381    proxy: &ProxyTarget,
382) -> Result<ProxiedStream, TransportError> {
383    let tcp = TcpStream::connect((proxy.host.as_str(), proxy.port))
384        .await
385        .map_err(TransportError::Io)?;
386
387    crate::net::apply_socket_options(&tcp);
388
389    if proxy.is_tls {
390        let proxy_tls = wrap_tls(tcp, &proxy.host).await?;
391        let tunneled = send_connect(proxy_tls, target, proxy).await?;
392        if target.is_tls {
393            let upstream = wrap_tls(tunneled, &target.host).await?;
394            Ok(ProxiedStream::TlsOverTlsProxy(Box::new(upstream)))
395        } else {
396            Ok(ProxiedStream::PlainOverTlsProxy(Box::new(tunneled)))
397        }
398    } else {
399        let tunneled = send_connect(tcp, target, proxy).await?;
400        if target.is_tls {
401            let upstream = wrap_tls(tunneled, &target.host).await?;
402            Ok(ProxiedStream::Tls(Box::new(upstream)))
403        } else {
404            Ok(ProxiedStream::Plain(tunneled))
405        }
406    }
407}
408
409/// Sends a `CONNECT` request and returns the underlying stream once a `2xx`
410/// status is received. The returned stream is positioned after the empty line
411/// terminating the proxy response headers.
412async fn send_connect<S>(
413    mut stream: S,
414    target: &WsTarget,
415    proxy: &ProxyTarget,
416) -> Result<S, TransportError>
417where
418    S: AsyncRead + AsyncWrite + Unpin,
419{
420    let host_header = format_host_header(&target.host, target.port);
421    let mut request = format!(
422        "CONNECT {host_header} HTTP/1.1\r\n\
423         Host: {host_header}\r\n\
424         Proxy-Connection: Keep-Alive\r\n"
425    );
426
427    if let Some(auth) = &proxy.auth_header {
428        request.push_str("Proxy-Authorization: ");
429        request.push_str(auth);
430        request.push_str("\r\n");
431    }
432    request.push_str("\r\n");
433
434    stream
435        .write_all(request.as_bytes())
436        .await
437        .map_err(TransportError::Io)?;
438    stream.flush().await.map_err(TransportError::Io)?;
439
440    read_connect_response(&mut stream).await?;
441    Ok(stream)
442}
443
444fn format_host_header(host: &str, port: u16) -> String {
445    if host.contains(':') && !(host.starts_with('[') && host.ends_with(']')) {
446        format!("[{host}]:{port}")
447    } else {
448        format!("{host}:{port}")
449    }
450}
451
452/// Reads the proxy's response up to the empty line that terminates the
453/// headers, validating the status line.
454async fn read_connect_response<S>(stream: &mut S) -> Result<(), TransportError>
455where
456    S: AsyncRead + Unpin,
457{
458    let mut buf = Vec::with_capacity(512);
459    let mut byte = [0u8; 1];
460
461    loop {
462        let n = stream.read(&mut byte).await.map_err(TransportError::Io)?;
463        if n == 0 {
464            return Err(TransportError::ConnectionClosed);
465        }
466
467        buf.push(byte[0]);
468
469        if buf.ends_with(b"\r\n\r\n") {
470            break;
471        }
472
473        if buf.len() > MAX_PROXY_RESPONSE_BYTES {
474            return Err(TransportError::Handshake(format!(
475                "proxy CONNECT response exceeded {MAX_PROXY_RESPONSE_BYTES} bytes without terminator"
476            )));
477        }
478    }
479
480    let text = std::str::from_utf8(&buf).map_err(|_| {
481        TransportError::Handshake("proxy CONNECT response was not valid UTF-8".to_string())
482    })?;
483
484    let status_line = text.lines().next().ok_or_else(|| {
485        TransportError::Handshake("proxy CONNECT response missing status line".to_string())
486    })?;
487
488    // Expect: `HTTP/1.1 200 Connection established` (or any 2xx).
489    let mut parts = status_line.splitn(3, ' ');
490    let version = parts.next().ok_or_else(|| {
491        TransportError::Handshake("proxy CONNECT response has a malformed status line".to_string())
492    })?;
493
494    // The version gates the status branch below: without this check a malformed line such as
495    // `NOT-HTTP 503 ...` would yield a retryable `ProxyConnectRejected` rather than staying a
496    // permanent handshake failure.
497    if !matches!(version, "HTTP/1.0" | "HTTP/1.1") {
498        return Err(TransportError::Handshake(
499            "proxy CONNECT response has a malformed status line".to_string(),
500        ));
501    }
502
503    // Parsed as a `StatusCode` rather than a bare `u16` for the same reason the version is
504    // validated above: the number now selects retry behavior, and `u16` parsing accepts tokens
505    // HTTP does not, normalizing `0503` to `503`.
506    let status_code = parts
507        .next()
508        .ok_or_else(|| {
509            TransportError::Handshake(
510                "proxy CONNECT response has a malformed status line".to_string(),
511            )
512        })?
513        .parse::<http::StatusCode>()
514        .map_err(|_| {
515            TransportError::Handshake(
516                "proxy CONNECT response has a invalid status code".to_string(),
517            )
518        })?
519        .as_u16();
520
521    if !(200..300).contains(&status_code) {
522        return Err(TransportError::ProxyConnectRejected(status_code));
523    }
524
525    Ok(())
526}
527
528/// Wraps a stream in a `rustls`-backed TLS session using `webpki_roots`.
529async fn wrap_tls<S>(stream: S, server_name: &str) -> Result<TlsStream<S>, TransportError>
530where
531    S: AsyncRead + AsyncWrite + Unpin,
532{
533    let mut root_store = RootCertStore::empty();
534    root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
535
536    let config = ClientConfig::builder()
537        .with_root_certificates(root_store)
538        .with_no_client_auth();
539
540    let connector = TlsConnector::from(std::sync::Arc::new(config));
541    let domain = ServerName::try_from(server_name.to_string())
542        .map_err(|e| TransportError::Tls(format!("invalid DNS name '{server_name}': {e}")))?;
543
544    connector
545        .connect(domain, stream)
546        .await
547        .map_err(TransportError::Io)
548}
549
550#[cfg(test)]
551#[cfg(not(feature = "turmoil"))] // proxy hop is not modeled under the turmoil simulator
552mod tests {
553    use std::net::SocketAddr;
554
555    use rstest::rstest;
556    use tokio::net::TcpListener;
557
558    use super::*;
559
560    #[rstest]
561    fn ws_target_parses_wss() {
562        let target = WsTarget::parse("wss://stream.binance.com:9443/ws/btcusdt@trade").unwrap();
563        assert_eq!(target.host, "stream.binance.com");
564        assert_eq!(target.port, 9443);
565        assert!(target.is_tls);
566    }
567
568    #[rstest]
569    fn ws_target_default_ports() {
570        let plain = WsTarget::parse("ws://example.com/path").unwrap();
571        assert_eq!(plain.port, 80);
572        assert!(!plain.is_tls);
573
574        let tls = WsTarget::parse("wss://example.com/path").unwrap();
575        assert_eq!(tls.port, 443);
576        assert!(tls.is_tls);
577    }
578
579    #[rstest]
580    fn ws_target_strips_ipv6_brackets() {
581        let target = WsTarget::parse("wss://[::1]:9443/ws").unwrap();
582        assert_eq!(target.host, "::1");
583        assert_eq!(target.port, 9443);
584    }
585
586    #[rstest]
587    fn ws_target_rejects_non_ws_scheme() {
588        let err = WsTarget::parse("https://example.com").unwrap_err();
589        assert!(matches!(err, TransportError::InvalidUrl(_)));
590    }
591
592    #[rstest]
593    fn proxy_target_parses_http() {
594        let proxy = ProxyTarget::parse("http://127.0.0.1:9999").unwrap();
595        assert_eq!(proxy.host, "127.0.0.1");
596        assert_eq!(proxy.port, 9999);
597        assert!(!proxy.is_tls);
598        assert!(proxy.auth_header.is_none());
599    }
600
601    #[rstest]
602    fn proxy_target_default_ports() {
603        let plain = ProxyTarget::parse("http://proxy.example.com").unwrap();
604        assert_eq!(plain.port, 80);
605        let tls = ProxyTarget::parse("https://proxy.example.com").unwrap();
606        assert_eq!(tls.port, 443);
607        assert!(tls.is_tls);
608    }
609
610    #[rstest]
611    fn proxy_target_basic_auth() {
612        let proxy =
613            ProxyTarget::parse("http://proxytest:fixture42@proxy.example.com:8080").unwrap();
614        let expected = format!("Basic {}", BASE64.encode("proxytest:fixture42"));
615        assert_eq!(proxy.auth_header.unwrap(), expected);
616    }
617
618    #[rstest]
619    fn proxy_target_basic_auth_decodes_percent_encoded() {
620        // `p%40ss` should decode to `p@ss` before assembling Basic credentials
621        let proxy = ProxyTarget::parse("http://us%2Fer:p%40ss@proxy.example.com:8080").unwrap();
622        let header = proxy.auth_header.unwrap();
623        let expected = format!("Basic {}", BASE64.encode("us/er:p@ss"));
624        assert_eq!(header, expected);
625    }
626
627    #[rstest]
628    fn proxy_target_basic_auth_with_empty_username() {
629        let proxy = ProxyTarget::parse("http://:fixture42@proxy.example.com:8080").unwrap();
630        let expected = format!("Basic {}", BASE64.encode(":fixture42"));
631
632        assert_eq!(proxy.auth_header.unwrap(), expected);
633    }
634
635    #[rstest]
636    fn proxy_debug_redacts_credentials() {
637        const SECRET: &str = "unique-proxy-secret";
638        let url = format!("http://proxytest:{SECRET}@proxy.example.com:8080");
639        let proxy_url = ProxyUrl::parse(url.clone()).unwrap();
640        let target = ProxyTarget::parse(&url).unwrap();
641        let proxy_url_debug = format!("{proxy_url:?}");
642        let target_debug = format!("{target:?}");
643        let encoded_credentials = BASE64.encode(format!("proxytest:{SECRET}"));
644
645        assert_eq!(proxy_url_debug, "ProxyUrl(\"<redacted>\")");
646        assert!(!proxy_url_debug.contains(SECRET));
647        assert!(!target_debug.contains(SECRET));
648        assert!(!target_debug.contains(&encoded_credentials));
649        assert!(target_debug.contains(REDACTED));
650    }
651
652    #[rstest]
653    fn proxy_parse_error_redacts_credentials() {
654        const SECRET: &str = "unique-proxy-secret";
655        let err = ProxyUrl::parse(format!("http://proxytest:{SECRET}@[::1"))
656            .expect_err("malformed proxy URL should fail");
657
658        assert!(!err.to_string().contains(SECRET));
659    }
660
661    #[tokio::test]
662    async fn send_connect_includes_proxy_authorization() {
663        let (client, mut server) = tokio::io::duplex(1024);
664        let target = WsTarget::parse("ws://example.com:80/path").unwrap();
665        let proxy =
666            ProxyTarget::parse("http://proxytest:fixture42@proxy.example.com:8080").unwrap();
667
668        let server_task = tokio::spawn(async move {
669            let mut request = Vec::new();
670            let mut chunk = [0; 64];
671            loop {
672                let n = server.read(&mut chunk).await.unwrap();
673                request.extend_from_slice(&chunk[..n]);
674                if request.ends_with(b"\r\n\r\n") {
675                    break;
676                }
677            }
678
679            let request = String::from_utf8(request).unwrap();
680            let expected = format!(
681                "CONNECT example.com:80 HTTP/1.1\r\n\
682                 Host: example.com:80\r\n\
683                 Proxy-Connection: Keep-Alive\r\n\
684                 Proxy-Authorization: Basic {}\r\n\r\n",
685                BASE64.encode("proxytest:fixture42")
686            );
687            assert_eq!(request, expected);
688            server
689                .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n")
690                .await
691        });
692
693        let _stream = send_connect(client, &target, &proxy).await.unwrap();
694        server_task.await.unwrap().unwrap();
695    }
696
697    #[rstest]
698    fn proxy_target_strips_ipv6_brackets() {
699        let proxy = ProxyTarget::parse("http://[::1]:8080").unwrap();
700        assert_eq!(proxy.host, "::1");
701        assert_eq!(proxy.port, 8080);
702    }
703
704    #[rstest]
705    fn proxy_target_rejects_socks() {
706        let err = ProxyTarget::parse("socks5://127.0.0.1:1080").unwrap_err();
707        let TransportError::InvalidUrl(msg) = err else {
708            panic!("expected InvalidUrl");
709        };
710        assert!(msg.contains("SOCKS"));
711    }
712
713    #[rstest]
714    fn proxy_kind_classifies_http() {
715        let kind = ProxyKind::parse("http://127.0.0.1:9999").unwrap();
716        assert!(matches!(kind, ProxyKind::Http(_)));
717    }
718
719    #[rstest]
720    fn proxy_kind_classifies_socks_as_unsupported() {
721        let kind = ProxyKind::parse("socks5://127.0.0.1:1080").unwrap();
722        let ProxyKind::Unsupported { scheme } = kind else {
723            panic!("expected Unsupported");
724        };
725        assert_eq!(scheme, "socks5");
726    }
727
728    #[rstest]
729    fn proxy_kind_rejects_garbage() {
730        assert!(ProxyKind::parse("ftp://x").is_err());
731        assert!(ProxyKind::parse("").is_err());
732    }
733
734    #[rstest]
735    fn proxy_kind_rejects_socks_without_authority() {
736        // `socks5:host:port` (no `//`) parses as scheme + opaque path; surface
737        // as a real error instead of a silent direct-fallback.
738        let err = ProxyKind::parse("socks5:127.0.0.1:1080").unwrap_err();
739        assert!(matches!(err, TransportError::InvalidUrl(_)));
740    }
741
742    #[rstest]
743    fn proxy_target_rejects_unknown_scheme() {
744        let err = ProxyTarget::parse("ftp://proxy.example.com").unwrap_err();
745        assert!(matches!(err, TransportError::InvalidUrl(_)));
746    }
747
748    #[rstest]
749    fn proxy_target_rejects_empty() {
750        let err = ProxyTarget::parse("").unwrap_err();
751        assert!(matches!(err, TransportError::InvalidUrl(_)));
752    }
753
754    #[rstest]
755    fn host_header_brackets_ipv6() {
756        assert_eq!(format_host_header("example.com", 443), "example.com:443");
757        assert_eq!(format_host_header("::1", 443), "[::1]:443");
758        assert_eq!(format_host_header("[::1]", 443), "[::1]:443");
759    }
760
761    /// Spawn a fake HTTP proxy that returns the configured response after
762    /// reading one CONNECT request line. Returns the bound address.
763    async fn spawn_fake_proxy(response: &'static [u8]) -> SocketAddr {
764        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
765        let addr = listener.local_addr().unwrap();
766        tokio::spawn(async move {
767            let (mut stream, _) = listener.accept().await.unwrap();
768            let mut buf = vec![0u8; 1024];
769            // Read until we see the CONNECT terminator.
770            loop {
771                let n = AsyncReadExt::read(&mut stream, &mut buf).await.unwrap();
772                if n == 0 {
773                    break;
774                }
775
776                if buf[..n].windows(4).any(|w| w == b"\r\n\r\n") {
777                    break;
778                }
779            }
780            stream.write_all(response).await.unwrap();
781            stream.flush().await.unwrap();
782        });
783        addr
784    }
785
786    #[tokio::test]
787    async fn read_connect_response_accepts_2xx() {
788        let addr = spawn_fake_proxy(b"HTTP/1.1 200 Connection established\r\n\r\n").await;
789        let mut stream = TcpStream::connect(addr).await.unwrap();
790        stream
791            .write_all(b"CONNECT host:443 HTTP/1.1\r\nHost: host:443\r\n\r\n")
792            .await
793            .unwrap();
794        stream.flush().await.unwrap();
795        read_connect_response(&mut stream).await.unwrap();
796    }
797
798    #[tokio::test]
799    async fn read_connect_response_rejects_403() {
800        let addr = spawn_fake_proxy(b"HTTP/1.1 403 Forbidden\r\n\r\n").await;
801        let mut stream = TcpStream::connect(addr).await.unwrap();
802        stream
803            .write_all(b"CONNECT host:443 HTTP/1.1\r\nHost: host:443\r\n\r\n")
804            .await
805            .unwrap();
806        stream.flush().await.unwrap();
807        let err = read_connect_response(&mut stream).await.unwrap_err();
808        assert!(matches!(err, TransportError::ProxyConnectRejected(403)));
809    }
810
811    /// 300 sits on the upper boundary of the accepted `200..300` range; if
812    /// the check is ever loosened to `200..=300` this test fails. 407 is the
813    /// classic "Proxy Authentication Required" response. Non-numeric status
814    /// probes the parse path.
815    #[rstest]
816    #[case::status_300(&b"HTTP/1.1 300 Multiple Choices\r\n\r\n"[..], Some(300), None)]
817    #[case::status_407(
818        &b"HTTP/1.1 407 Proxy Authentication Required\r\nProxy-Authenticate: Basic\r\n\r\n"[..],
819        Some(407),
820        None,
821    )]
822    #[case::malformed_status(
823        &b"HTTP/1.1 abc Boom\r\n\r\n"[..],
824        None,
825        Some("invalid status code"),
826    )]
827    #[tokio::test]
828    async fn read_connect_response_rejects_non_2xx(
829        #[case] response: &'static [u8],
830        #[case] expected_status: Option<u16>,
831        #[case] expected_msg_substring: Option<&'static str>,
832    ) {
833        let addr = spawn_fake_proxy(response).await;
834        let mut stream = TcpStream::connect(addr).await.unwrap();
835        stream
836            .write_all(b"CONNECT host:443 HTTP/1.1\r\nHost: host:443\r\n\r\n")
837            .await
838            .unwrap();
839        stream.flush().await.unwrap();
840        let err = read_connect_response(&mut stream).await.unwrap_err();
841
842        match (expected_status, expected_msg_substring) {
843            (Some(status), None) => {
844                assert!(
845                    matches!(err, TransportError::ProxyConnectRejected(actual) if actual == status)
846                );
847            }
848            (None, Some(expected_msg_substring)) => {
849                let TransportError::Handshake(msg) = err else {
850                    panic!("expected Handshake error, was {err:?}");
851                };
852                assert!(
853                    msg.contains(expected_msg_substring),
854                    "expected error message to contain {expected_msg_substring:?}, was {msg:?}"
855                );
856            }
857            _ => unreachable!(),
858        }
859    }
860
861    #[tokio::test]
862    async fn read_connect_response_does_not_expose_reason_phrase() {
863        const SECRET: &str = "unique-proxy-secret";
864        let response = format!("HTTP/1.1 407 {SECRET}\r\n\r\n").into_bytes();
865        let response = Box::leak(response.into_boxed_slice());
866        let addr = spawn_fake_proxy(response).await;
867        let mut stream = TcpStream::connect(addr).await.unwrap();
868        stream
869            .write_all(b"CONNECT host:443 HTTP/1.1\r\nHost: host:443\r\n\r\n")
870            .await
871            .unwrap();
872        stream.flush().await.unwrap();
873
874        let err = read_connect_response(&mut stream).await.unwrap_err();
875
876        assert_eq!(err.to_string(), "proxy CONNECT rejected with status 407");
877        assert!(!err.to_string().contains(SECRET));
878    }
879
880    /// Closing the connection mid-response should produce a clear handshake
881    /// error rather than spinning on a zero-byte read.
882    #[tokio::test]
883    async fn read_connect_response_rejects_eof_before_terminator() {
884        // Truncated response: missing the empty line that ends the headers
885        let addr = spawn_fake_proxy(b"HTTP/1.1 200 OK\r\n").await;
886        let mut stream = TcpStream::connect(addr).await.unwrap();
887        stream
888            .write_all(b"CONNECT host:443 HTTP/1.1\r\nHost: host:443\r\n\r\n")
889            .await
890            .unwrap();
891        stream.flush().await.unwrap();
892        let err = read_connect_response(&mut stream).await.unwrap_err();
893        assert!(matches!(err, TransportError::ConnectionClosed));
894    }
895
896    /// A malformed version with an otherwise retryable status must stay a permanent
897    /// `Handshake` failure. The status is 503 deliberately: were the version left
898    /// unvalidated, this would parse as `ProxyConnectRejected(503)` and be retried.
899    #[tokio::test]
900    async fn read_connect_response_rejects_malformed_version_with_retryable_status() {
901        let addr = spawn_fake_proxy(b"NOT-HTTP 503 Service Unavailable\r\n\r\n").await;
902        let mut stream = TcpStream::connect(addr).await.unwrap();
903        stream
904            .write_all(b"CONNECT host:443 HTTP/1.1\r\nHost: host:443\r\n\r\n")
905            .await
906            .unwrap();
907        stream.flush().await.unwrap();
908        let err = read_connect_response(&mut stream).await.unwrap_err();
909        let TransportError::Handshake(msg) = err else {
910            panic!("expected Handshake error, was {err:?}");
911        };
912        assert!(msg.contains("malformed status line"), "was {msg}");
913    }
914
915    /// `0503` is not a valid status token but parses as `503` under bare `u16`
916    /// parsing, which would make a malformed line retryable. 503 is used because
917    /// a permanent status would pass whether or not the token is validated.
918    #[tokio::test]
919    async fn read_connect_response_rejects_malformed_status_token_with_retryable_value() {
920        let addr = spawn_fake_proxy(b"HTTP/1.1 0503 Service Unavailable\r\n\r\n").await;
921        let mut stream = TcpStream::connect(addr).await.unwrap();
922        stream
923            .write_all(b"CONNECT host:443 HTTP/1.1\r\nHost: host:443\r\n\r\n")
924            .await
925            .unwrap();
926        stream.flush().await.unwrap();
927        let err = read_connect_response(&mut stream).await.unwrap_err();
928        let TransportError::Handshake(msg) = err else {
929            panic!("expected Handshake error, was {err:?}");
930        };
931        assert!(msg.contains("invalid status code"), "was {msg}");
932    }
933
934    /// A proxy that streams headers without ever emitting `\r\n\r\n` should
935    /// trip the size cap rather than allocating without bound.
936    #[tokio::test]
937    async fn read_connect_response_rejects_oversize_headers() {
938        let mut response = b"HTTP/1.1 200 OK\r\n".to_vec();
939        while response.len() <= MAX_PROXY_RESPONSE_BYTES {
940            response.extend_from_slice(b"X-Pad: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\r\n");
941        }
942        let leaked: &'static [u8] = response.leak();
943        let addr = spawn_fake_proxy(leaked).await;
944        let mut stream = TcpStream::connect(addr).await.unwrap();
945        stream
946            .write_all(b"CONNECT host:443 HTTP/1.1\r\nHost: host:443\r\n\r\n")
947            .await
948            .unwrap();
949        stream.flush().await.unwrap();
950        let err = read_connect_response(&mut stream).await.unwrap_err();
951        let TransportError::Handshake(msg) = err else {
952            panic!("expected Handshake error, was {err:?}");
953        };
954        assert!(
955            msg.contains("exceeded"),
956            "unexpected handshake error: {msg}"
957        );
958    }
959
960    /// After accepting the 2xx response, the stream cursor must sit immediately
961    /// after the terminating `\r\n\r\n` so the WebSocket handshake can read its
962    /// own response. Regression guard against over-reading the terminator.
963    #[tokio::test]
964    async fn read_connect_response_preserves_trailing_bytes() {
965        let addr = spawn_fake_proxy(b"HTTP/1.1 200 Connection established\r\n\r\nLEFTOVER").await;
966        let mut stream = TcpStream::connect(addr).await.unwrap();
967        stream
968            .write_all(b"CONNECT host:443 HTTP/1.1\r\nHost: host:443\r\n\r\n")
969            .await
970            .unwrap();
971        stream.flush().await.unwrap();
972        read_connect_response(&mut stream).await.unwrap();
973
974        let mut tail = [0u8; b"LEFTOVER".len()];
975        AsyncReadExt::read_exact(&mut stream, &mut tail)
976            .await
977            .unwrap();
978        assert_eq!(&tail, b"LEFTOVER");
979    }
980}
nautilus_network/websocket/proxy.rs

nautilus_network/websocket/
proxy.rs
