diff --git a/Cargo.lock b/Cargo.lock index 3b60dc4579..f30266bd6b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -769,6 +769,7 @@ dependencies = [ "buzz-core", "buzz-persona", "buzz-sdk", + "buzz-ws-client", "chrono", "clap", "evalexpr", @@ -1042,6 +1043,7 @@ name = "buzz-pairing-cli" version = "0.1.0" dependencies = [ "buzz-core", + "buzz-ws-client", "clap", "futures-util", "hex", @@ -1290,13 +1292,16 @@ name = "buzz-ws-client" version = "0.1.0" dependencies = [ "futures-util", + "hyper-util", "nostr", "serde_json", "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", + "tower-service", "tracing", "url", + "windows-registry", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 3ac7ee4cce..80bea9b127 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -110,8 +110,14 @@ moka = { version = "0.12", features = ["sync"] } # Async stream utilities futures-util = "0.3" -# WebSocket client (test client) +# WebSocket clients tokio-tungstenite = { version = "0.29", features = ["rustls-tls-webpki-roots"] } +hyper-util = { version = "0.1", features = [ + "client-legacy", + "client-proxy", + "client-proxy-system", +] } +tower-service = "0.3" url = "2" # Property-based testing (dev-only) diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806..f616c79fc7 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -19,6 +19,7 @@ path = "src/main.rs" # Internal buzz-core = { workspace = true } buzz-sdk = { workspace = true } +buzz-ws-client = { workspace = true } buzz-persona = { path = "../buzz-persona" } # Nostr diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index e6164b02dd..2c0190b2a9 100644 --- a/crates/buzz-acp/README.md +++ b/crates/buzz-acp/README.md @@ -115,6 +115,11 @@ All configuration is via environment variables (or CLI flags — every env var h | `BUZZ_ACP_MAX_TURN_DURATION` | no | `7200` | Absolute wall-clock cap per turn (safety valve). | | `BUZZ_API_TOKEN` | no | — | API token (required if relay enforces token auth). | +Relay WebSocket connections honor `HTTP_PROXY`, `HTTPS_PROXY`, `ALL_PROXY`, +and `NO_PROXY`. Static macOS and Windows system proxy settings are used when +the environment does not override them. HTTP CONNECT and SOCKS proxy URLs are +supported; loopback relay URLs always connect directly. + **Note:** `BUZZ_ACP_AGENT_ARGS` splits on commas. For args with values, use: `-c,key="value"`. **Legacy env vars:** `BUZZ_ACP_PRIVATE_KEY`, `BUZZ_ACP_API_TOKEN`, and `BUZZ_ACP_TURN_TIMEOUT` (replaced by `BUZZ_ACP_IDLE_TIMEOUT`) are still accepted as fallbacks. diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index c8312cc61e..fe6b008965 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -117,12 +117,13 @@ use buzz_core::kind::{ KIND_AGENT_OBSERVER_FRAME, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_TYPING_INDICATOR, }; +use buzz_ws_client::connect_websocket; use futures_util::{SinkExt, StreamExt}; use nostr::{Event, EventBuilder, Keys, Kind, RelayUrl, Tag}; use serde_json::{json, Value}; use tokio::sync::mpsc; use tokio::time::timeout; -use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream}; +use tokio_tungstenite::{tungstenite::Message, MaybeTlsStream, WebSocketStream}; use tracing::{debug, info, warn}; use uuid::Uuid; @@ -436,7 +437,7 @@ pub struct BuzzEvent { #[derive(Debug, thiserror::Error)] pub enum RelayError { #[error("WebSocket error: {0}")] - WebSocket(Box), + WebSocket(#[source] Box), #[error("JSON error: {0}")] Json(#[from] serde_json::Error), @@ -3346,17 +3347,27 @@ fn jittered_duration(base: Duration) -> Duration { /// Classify a `RelayError` as a DNS resolution failure. /// -/// Matches the OS-level "name not found" strings surfaced by the platform's -/// resolver, covering macOS (`nodename nor servname`), Linux (`Name or service not -/// known`), and common BSD/Windows variants (`No such host`, -/// `failed to lookup address`). These are transient on brownouts and must NOT -/// consume a backoff ladder rung — they retry on a flat `DNS_RETRY_INTERVAL`. +/// Walks the error source chain so connector context does not hide the +/// resolver failure. Matches hyper-util's typed DNS error plus the OS-level +/// strings surfaced on macOS, Linux, BSD, and Windows. These are transient on +/// brownouts and must NOT consume a backoff ladder rung — they retry on a flat +/// `DNS_RETRY_INTERVAL`. pub(crate) fn is_dns_error(err: &RelayError) -> bool { - let msg = err.to_string(); - msg.contains("nodename nor servname") - || msg.contains("Name or service not known") - || msg.contains("No such host") - || msg.contains("failed to lookup address") + let mut current: Option<&(dyn std::error::Error + 'static)> = Some(err); + while let Some(error) = current { + let message = error.to_string().to_ascii_lowercase(); + if message == "dns error" + || message.contains("nodename nor servname") + || message.contains("name or service not known") + || message.contains("no such host") + || message.contains("failed to lookup address") + { + return true; + } + current = error.source(); + } + + false } /// Shutdown-aware fixed-duration sleep for REQ pacing in `resubscribe_after_reconnect`. @@ -3652,7 +3663,7 @@ pub(crate) fn parse_relay_message(text: &str) -> Result bool { match err { @@ -3708,7 +3719,7 @@ fn is_terminal_ws_error(err: &tokio_tungstenite::tungstenite::Error) -> bool { // downcast above). WsError::Tls(_) => true, - // Unreachable during connect_async; kept fail-safe transient. + // Unreachable during connection establishment; kept fail-safe transient. WsError::AlreadyClosed | WsError::WriteBufferFull(_) => false, } } @@ -3831,7 +3842,7 @@ async fn do_connect( .parse::() .map_err(|e| RelayError::Http(format!("invalid relay URL: {e}")))?; - let (ws, _response) = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(parsed.as_str())) + let (ws, _response) = tokio::time::timeout(CONNECT_TIMEOUT, connect_websocket(parsed.as_str())) .await .map_err(|_| RelayError::ConnectionClosed)? // timeout → treat as connection failure .map_err(|e| RelayError::WebSocket(Box::new(e)))?; @@ -4348,7 +4359,7 @@ mod tests { .await .expect("complete server websocket handshake") }); - let (client, _) = connect_async(format!("ws://{address}")) + let (client, _) = tokio_tungstenite::connect_async(format!("ws://{address}")) .await .expect("connect test websocket"); (client, server.await.expect("join test websocket server")) @@ -6013,7 +6024,7 @@ mod tests { "failed to lookup address information".into() ))); // F15: production-shaped error — RelayError::WebSocket wrapping a - // tungstenite I/O error (the shape emitted by connect_async on macOS). + // tungstenite I/O error (the shape emitted by connection failures on macOS). let ws_io_err = RelayError::WebSocket(Box::new(tungstenite::Error::Io( std::io::Error::other("nodename nor servname provided, or not known"), ))); @@ -6021,6 +6032,21 @@ mod tests { is_dns_error(&ws_io_err), "WebSocket-wrapped I/O DNS error must be classified as DNS" ); + #[derive(Debug, thiserror::Error)] + #[error("connection failed")] + struct NestedConnectionError { + #[source] + source: std::io::Error, + } + let nested_dns_err = RelayError::WebSocket(Box::new(tungstenite::Error::Io( + std::io::Error::other(NestedConnectionError { + source: std::io::Error::other("Name or service not known"), + }), + ))); + assert!( + is_dns_error(&nested_dns_err), + "connector context must not hide a nested resolver failure" + ); // Normal connection errors are NOT DNS errors. assert!(!is_dns_error(&RelayError::Timeout)); assert!(!is_dns_error(&RelayError::ConnectionClosed)); diff --git a/crates/buzz-pairing-cli/Cargo.toml b/crates/buzz-pairing-cli/Cargo.toml index 58fefe32d9..4f432fbb83 100644 --- a/crates/buzz-pairing-cli/Cargo.toml +++ b/crates/buzz-pairing-cli/Cargo.toml @@ -13,6 +13,7 @@ path = "src/main.rs" [dependencies] buzz-core = { workspace = true } +buzz-ws-client = { workspace = true } nostr = { workspace = true } tokio = { workspace = true } tokio-tungstenite = { workspace = true } diff --git a/crates/buzz-pairing-cli/src/main.rs b/crates/buzz-pairing-cli/src/main.rs index 1eb9d215f9..553421c004 100644 --- a/crates/buzz-pairing-cli/src/main.rs +++ b/crates/buzz-pairing-cli/src/main.rs @@ -23,11 +23,12 @@ use buzz_core::pairing::{ types::PayloadType, PairingError, }; +use buzz_ws_client::connect_websocket; use clap::{Parser, Subcommand}; use futures_util::{SinkExt, StreamExt}; use nostr::{Event, EventBuilder, Keys, RelayUrl, SecretKey, ToBech32}; use tokio::time::timeout; -use tokio_tungstenite::{connect_async, tungstenite::Message}; +use tokio_tungstenite::tungstenite::Message; use zeroize::Zeroizing; #[derive(Parser)] @@ -125,7 +126,7 @@ async fn cmd_source(relay_url: String, nsec: Option) -> Result<(), CliEr // Connect to relay and handle NIP-42 auth if required. // Auth uses the session's ephemeral keys so the relay accepts our events. - let (ws, _) = connect_async(&relay_url).await?; + let (ws, _) = connect_websocket(&relay_url).await?; let (mut write, mut read) = ws.split(); handle_nip42_auth(&mut read, &mut write, &session, &relay_url).await?; @@ -227,7 +228,7 @@ async fn cmd_target(relay_override: Option, show_secret: bool) -> Result let (mut session, offer_event) = PairingSession::new_target(&qr)?; // Connect to relay and handle NIP-42 auth if required. - let (ws, _) = connect_async(&relay_url).await?; + let (ws, _) = connect_websocket(&relay_url).await?; let (mut write, mut read) = ws.split(); handle_nip42_auth(&mut read, &mut write, &session, &relay_url).await?; diff --git a/crates/buzz-ws-client/Cargo.toml b/crates/buzz-ws-client/Cargo.toml index 5cec925677..2bb835edc8 100644 --- a/crates/buzz-ws-client/Cargo.toml +++ b/crates/buzz-ws-client/Cargo.toml @@ -10,8 +10,13 @@ repository.workspace = true nostr = { workspace = true } tokio = { workspace = true } tokio-tungstenite = { workspace = true } +hyper-util = { workspace = true } +tower-service = { workspace = true } futures-util = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } url = { workspace = true } tracing = { workspace = true } + +[target.'cfg(windows)'.dependencies] +windows-registry = "0.6" diff --git a/crates/buzz-ws-client/src/connection.rs b/crates/buzz-ws-client/src/connection.rs index bec5b56bb4..15e750fe70 100644 --- a/crates/buzz-ws-client/src/connection.rs +++ b/crates/buzz-ws-client/src/connection.rs @@ -5,11 +5,12 @@ use futures_util::{SinkExt, StreamExt}; use nostr::{Event, Keys, Tag}; use serde_json::{json, Value}; use tokio::time::timeout; -use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream}; +use tokio_tungstenite::{tungstenite::Message, MaybeTlsStream, WebSocketStream}; use tracing::debug; use crate::error::WsClientError; use crate::message::{build_auth_event, parse_relay_message, OkResponse, RelayMessage}; +use crate::proxy::connect_websocket; type WsStream = WebSocketStream>; @@ -50,7 +51,7 @@ impl NostrWsConnection { .parse::() .map_err(|e| WsClientError::Url(e.to_string()))?; - let (ws, _response) = connect_async(parsed.as_str()) + let (ws, _response) = connect_websocket(parsed.as_str()) .await .map_err(WsClientError::WebSocket)?; diff --git a/crates/buzz-ws-client/src/lib.rs b/crates/buzz-ws-client/src/lib.rs index 02c7d4b1b2..23a64409cd 100644 --- a/crates/buzz-ws-client/src/lib.rs +++ b/crates/buzz-ws-client/src/lib.rs @@ -3,7 +3,9 @@ pub mod connection; pub mod error; pub mod message; +pub mod proxy; pub use connection::{publish_event, NostrWsConnection}; pub use error::WsClientError; pub use message::{build_auth_event, parse_relay_message, OkResponse, RelayMessage}; +pub use proxy::connect_websocket; diff --git a/crates/buzz-ws-client/src/proxy/mod.rs b/crates/buzz-ws-client/src/proxy/mod.rs new file mode 100644 index 0000000000..e3eff06f76 --- /dev/null +++ b/crates/buzz-ws-client/src/proxy/mod.rs @@ -0,0 +1,434 @@ +//! Proxy-aware WebSocket connection establishment. + +use std::error::Error as StdError; +use std::io; + +use hyper_util::client::legacy::connect::proxy::{SocksV4, SocksV5, Tunnel}; +use hyper_util::client::legacy::connect::HttpConnector; +use hyper_util::client::proxy::matcher::{Intercept, Matcher}; +use tokio::net::TcpStream; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::error::UrlError; +use tokio_tungstenite::tungstenite::handshake::client::{Request, Response}; +use tokio_tungstenite::tungstenite::http::{uri::Authority, Uri}; +use tokio_tungstenite::tungstenite::Error; +use tokio_tungstenite::{client_async_tls, MaybeTlsStream, WebSocketStream}; +use tower_service::Service; + +#[cfg(any(windows, test))] +mod windows; + +/// A WebSocket stream whose TCP connection may run through a configured proxy. +pub type ProxyWebSocketStream = WebSocketStream>; + +/// Connect to a WebSocket using environment and platform proxy settings. +/// +/// `HTTP_PROXY`, `HTTPS_PROXY`, `ALL_PROXY`, and `NO_PROXY` are honored on all +/// platforms. Static system proxy settings are also discovered on macOS and +/// Windows. Loopback destinations always connect directly so a system proxy +/// cannot break Buzz's default local relay. +/// +/// HTTP proxies use CONNECT tunneling. SOCKS4, SOCKS4a, SOCKS5, and SOCKS5h +/// proxy URLs are supported. +pub async fn connect_websocket(request: R) -> Result<(ProxyWebSocketStream, Response), Error> +where + R: IntoClientRequest + Unpin, +{ + let request = request.into_client_request()?; + let matcher = system_proxy_matcher(); + connect_websocket_with(&matcher, request).await +} + +fn system_proxy_matcher() -> Matcher { + #[cfg(windows)] + { + windows::system_proxy_matcher() + } + + #[cfg(not(windows))] + { + Matcher::from_system() + } +} + +async fn connect_websocket_with( + matcher: &Matcher, + request: Request, +) -> Result<(ProxyWebSocketStream, Response), Error> { + let target = proxy_target_uri(request.uri())?; + let stream = connect_tcp(matcher, &target).await?; + + client_async_tls(request, stream).await +} + +async fn connect_tcp(matcher: &Matcher, target: &Uri) -> Result { + if target.host().is_some_and(is_loopback_host) { + return connect_direct(target).await; + } + + match matcher.intercept(target) { + Some(proxy) => connect_via_proxy(target, proxy).await, + None => connect_direct(target).await, + } +} + +async fn connect_direct(target: &Uri) -> Result { + let mut connector = tcp_connector(); + connector + .call(target.clone()) + .await + .map(|stream| stream.into_inner()) + .map_err(|error| transport_error("direct TCP connection failed", error)) +} + +async fn connect_via_proxy(target: &Uri, proxy: Intercept) -> Result { + let scheme = proxy.uri().scheme_str().unwrap_or("http"); + + match scheme { + "http" => connect_http_proxy(target, &proxy).await, + "socks4" | "socks4a" => connect_socks4_proxy(target, &proxy, scheme).await, + "socks5" | "socks5h" => connect_socks5_proxy(target, &proxy, scheme).await, + unsupported => Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("unsupported WebSocket proxy scheme: {unsupported}"), + ) + .into()), + } +} + +async fn connect_http_proxy(target: &Uri, proxy: &Intercept) -> Result { + let proxy_uri = proxy_tcp_uri(proxy.uri(), 80)?; + let mut connector = Tunnel::new(proxy_uri, tcp_connector()); + if let Some(auth) = proxy.basic_auth() { + connector = connector.with_auth(auth.clone()); + } + + connector + .call(target.clone()) + .await + .map(|stream| stream.into_inner()) + .map_err(|error| transport_error("HTTP proxy tunnel failed", error)) +} + +async fn connect_socks4_proxy( + target: &Uri, + proxy: &Intercept, + scheme: &str, +) -> Result { + if proxy.raw_auth().is_some() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "SOCKS4 proxy authentication is not supported", + ) + .into()); + } + + let proxy_uri = proxy_tcp_uri(proxy.uri(), 1080)?; + let mut connector = SocksV4::new(proxy_uri, tcp_connector()).local_dns(scheme == "socks4"); + + connector + .call(target.clone()) + .await + .map(|stream| stream.into_inner()) + .map_err(|error| transport_error("SOCKS4 proxy tunnel failed", error)) +} + +async fn connect_socks5_proxy( + target: &Uri, + proxy: &Intercept, + scheme: &str, +) -> Result { + let proxy_uri = proxy_tcp_uri(proxy.uri(), 1080)?; + let mut connector = SocksV5::new(proxy_uri, tcp_connector()).local_dns(scheme == "socks5"); + if let Some((username, password)) = proxy.raw_auth() { + connector = connector.with_auth(username.to_string(), password.to_string()); + } + + connector + .call(target.clone()) + .await + .map(|stream| stream.into_inner()) + .map_err(|error| transport_error("SOCKS5 proxy tunnel failed", error)) +} + +fn tcp_connector() -> HttpConnector { + let mut connector = HttpConnector::new(); + connector.enforce_http(false); + connector +} + +fn proxy_target_uri(websocket_uri: &Uri) -> Result { + let (scheme, default_port) = match websocket_uri.scheme_str() { + Some("ws") => ("http", 80), + Some("wss") => ("https", 443), + _ => return Err(UrlError::UnsupportedUrlScheme.into()), + }; + let authority = authority_with_port(websocket_uri, default_port)?; + let path_and_query = websocket_uri + .path_and_query() + .cloned() + .ok_or(UrlError::NoPathOrQuery)?; + + Uri::builder() + .scheme(scheme) + .authority(authority) + .path_and_query(path_and_query) + .build() + .map_err(Error::from) +} + +fn proxy_tcp_uri(proxy_uri: &Uri, default_port: u16) -> Result { + let authority = authority_with_port(proxy_uri, default_port)?; + Uri::builder() + .scheme("http") + .authority(authority) + .path_and_query("/") + .build() + .map_err(Error::from) +} + +fn authority_with_port(uri: &Uri, default_port: u16) -> Result { + let host = uri.host().ok_or(UrlError::NoHostName)?; + if host.is_empty() { + return Err(UrlError::EmptyHostName.into()); + } + + if let Some(port) = uri.port_u16() { + return format_authority(host, port); + } + + format_authority(host, default_port) +} + +fn format_authority(host: &str, port: u16) -> Result { + let host = if host.contains(':') && !host.starts_with('[') { + format!("[{host}]") + } else { + host.to_string() + }; + format!("{host}:{port}") + .parse::() + .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error).into()) +} + +fn is_loopback_host(host: &str) -> bool { + let host = host.trim_matches(['[', ']']); + let host = host.strip_suffix('.').unwrap_or(host); + host.eq_ignore_ascii_case("localhost") + || host.rsplit_once('.').is_some_and(|(prefix, suffix)| { + !prefix.is_empty() && suffix.eq_ignore_ascii_case("localhost") + }) + || host + .parse::() + .is_ok_and(|address| address.is_loopback()) +} + +fn transport_error( + operation: &'static str, + source: impl StdError + Send + Sync + 'static, +) -> Error { + io::Error::other(TransportError { + operation, + source: Box::new(source), + }) + .into() +} + +#[derive(Debug)] +struct TransportError { + operation: &'static str, + source: Box, +} + +impl std::fmt::Display for TransportError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "{}: {}", self.operation, self.source) + } +} + +impl StdError for TransportError { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(self.source.as_ref()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use futures_util::{SinkExt, StreamExt}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + use tokio_tungstenite::tungstenite::Message; + + #[test] + fn loopback_destinations_bypass_proxy() { + assert!(is_loopback_host("localhost")); + assert!(is_loopback_host("relay.localhost")); + assert!(is_loopback_host("relay.LOCALHOST")); + assert!(is_loopback_host("localhost.")); + assert!(is_loopback_host("127.0.0.1")); + assert!(is_loopback_host("[::1]")); + assert!(!is_loopback_host("relay.example.com")); + assert!(!is_loopback_host("127.0.0.2.example.com")); + } + + #[tokio::test] + async fn loopback_destination_connects_directly() { + let target_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let target_address = target_listener.local_addr().unwrap(); + let target_task = tokio::spawn(async move { + let (stream, _) = target_listener.accept().await.unwrap(); + tokio_tungstenite::accept_async(stream).await.unwrap() + }); + + let proxy_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy_address = proxy_listener.local_addr().unwrap(); + let matcher = Matcher::builder() + .all(format!("http://{proxy_address}")) + .build(); + let request = format!("ws://{target_address}") + .into_client_request() + .unwrap(); + + let (_websocket, response) = tokio::time::timeout( + std::time::Duration::from_secs(1), + connect_websocket_with(&matcher, request), + ) + .await + .expect("loopback connection must complete without waiting for the proxy") + .unwrap(); + + assert_eq!(response.status(), 101); + assert!( + tokio::time::timeout( + std::time::Duration::from_millis(50), + proxy_listener.accept() + ) + .await + .is_err(), + "loopback connection must not reach the proxy" + ); + target_task.await.unwrap(); + } + + #[tokio::test] + async fn connects_through_authenticated_http_proxy_without_target_dns() { + let proxy_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy_address = proxy_listener.local_addr().unwrap(); + + let proxy_task = tokio::spawn(async move { + let (mut stream, _) = proxy_listener.accept().await.unwrap(); + let mut request = Vec::new(); + loop { + let mut byte = [0_u8; 1]; + stream.read_exact(&mut byte).await.unwrap(); + request.push(byte[0]); + if request.ends_with(b"\r\n\r\n") { + break; + } + } + + let request = String::from_utf8(request).unwrap(); + assert!(request + .starts_with("CONNECT relay.invalid:443 HTTP/1.1\r\nHost: relay.invalid:443\r\n")); + assert!(request.contains("\r\nProxy-Authorization: Basic dXNlcjpwYXNzd29yZA==\r\n")); + + stream + .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n") + .await + .unwrap(); + + let mut websocket = tokio_tungstenite::accept_async(stream).await.unwrap(); + websocket + .send(Message::Text("proxied".into())) + .await + .unwrap(); + }); + + let matcher = Matcher::builder() + .all(format!("http://user:password@{proxy_address}")) + .build(); + let request = "ws://relay.invalid:443".into_client_request().unwrap(); + let (mut websocket, response) = connect_websocket_with(&matcher, request).await.unwrap(); + + assert_eq!(response.status(), 101); + assert_eq!( + websocket.next().await.unwrap().unwrap(), + Message::Text("proxied".into()) + ); + proxy_task.await.unwrap(); + } + + #[tokio::test] + async fn proxy_connection_errors_do_not_expose_credentials() { + let proxy_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy_address = proxy_listener.local_addr().unwrap(); + let proxy_task = tokio::spawn(async move { + let (_stream, _) = proxy_listener.accept().await.unwrap(); + }); + + let matcher = Matcher::builder() + .all(format!( + "http://sensitive-user:sensitive-password@{proxy_address}" + )) + .build(); + let request = "ws://relay.invalid:443".into_client_request().unwrap(); + let error = connect_websocket_with(&matcher, request) + .await + .unwrap_err() + .to_string(); + + assert!(!error.contains("sensitive-user")); + assert!(!error.contains("sensitive-password")); + proxy_task.await.unwrap(); + } + + #[tokio::test] + async fn connects_through_socks5h_proxy_without_target_dns() { + let proxy_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy_address = proxy_listener.local_addr().unwrap(); + + let proxy_task = tokio::spawn(async move { + let (mut stream, _) = proxy_listener.accept().await.unwrap(); + + let mut negotiation = [0_u8; 3]; + stream.read_exact(&mut negotiation).await.unwrap(); + assert_eq!(negotiation, [5, 1, 0]); + stream.write_all(&[5, 0]).await.unwrap(); + + let mut request_prefix = [0_u8; 5]; + stream.read_exact(&mut request_prefix).await.unwrap(); + assert_eq!(&request_prefix[..4], &[5, 1, 0, 3]); + + let mut target = vec![0_u8; usize::from(request_prefix[4]) + 2]; + stream.read_exact(&mut target).await.unwrap(); + let (host, port) = target.split_at(target.len() - 2); + assert_eq!(host, b"relay.invalid"); + assert_eq!(u16::from_be_bytes([port[0], port[1]]), 443); + + stream + .write_all(&[5, 0, 0, 1, 0, 0, 0, 0, 0, 0]) + .await + .unwrap(); + + let mut websocket = tokio_tungstenite::accept_async(stream).await.unwrap(); + websocket + .send(Message::Text("proxied".into())) + .await + .unwrap(); + }); + + let matcher = Matcher::builder() + .all(format!("socks5h://{proxy_address}")) + .build(); + let request = "ws://relay.invalid:443".into_client_request().unwrap(); + let (mut websocket, response) = connect_websocket_with(&matcher, request).await.unwrap(); + + assert_eq!(response.status(), 101); + assert_eq!( + websocket.next().await.unwrap().unwrap(), + Message::Text("proxied".into()) + ); + proxy_task.await.unwrap(); + } +} diff --git a/crates/buzz-ws-client/src/proxy/windows.rs b/crates/buzz-ws-client/src/proxy/windows.rs new file mode 100644 index 0000000000..b1a9264a68 --- /dev/null +++ b/crates/buzz-ws-client/src/proxy/windows.rs @@ -0,0 +1,158 @@ +//! Windows static proxy discovery. +//! +//! WinINet allows `ProxyServer` to contain protocol-specific entries such as +//! `http=proxy:8080;https=proxy:8443`. Hyper-util currently treats the entire +//! registry value as one URI, so Buzz normalizes the list before building the +//! shared matcher. + +#[cfg(windows)] +use hyper_util::client::proxy::matcher::Matcher; + +#[cfg(windows)] +const INTERNET_SETTINGS_KEY: &str = + "Software\\Microsoft\\Windows\\CurrentVersion\\Internet Settings"; + +#[cfg(windows)] +pub(super) fn system_proxy_matcher() -> Matcher { + if std::env::var_os("REQUEST_METHOD").is_some() { + return Matcher::from_env(); + } + + let mut http = first_env(&["HTTP_PROXY", "http_proxy"]); + let mut https = first_env(&["HTTPS_PROXY", "https_proxy"]); + let all = first_env(&["ALL_PROXY", "all_proxy"]); + let mut no = first_env(&["NO_PROXY", "no_proxy"]); + + if let Some(system) = read_system_proxy() { + if http.is_empty() && all.is_empty() { + http = system.http.unwrap_or_default(); + } + if https.is_empty() && all.is_empty() { + https = system.https.unwrap_or_default(); + } + if no.is_empty() { + no = system.no; + } + } + + Matcher::builder() + .all(all) + .http(http) + .https(https) + .no(no) + .build() +} + +#[cfg(windows)] +fn first_env(names: &[&str]) -> String { + names + .iter() + .find_map(|name| std::env::var(name).ok()) + .unwrap_or_default() +} + +#[cfg(windows)] +fn read_system_proxy() -> Option { + let settings = windows_registry::CURRENT_USER + .open(INTERNET_SETTINGS_KEY) + .ok()?; + if settings.get_u32("ProxyEnable").unwrap_or(0) == 0 { + return None; + } + + let mut proxy = settings + .get_string("ProxyServer") + .ok() + .map(|value| parse_proxy_server(&value)) + .unwrap_or_default(); + proxy.no = settings + .get_string("ProxyOverride") + .ok() + .map(|value| normalize_proxy_override(&value)) + .unwrap_or_default(); + Some(proxy) +} + +#[derive(Debug, Default, PartialEq, Eq)] +struct SystemProxy { + http: Option, + https: Option, + no: String, +} + +fn parse_proxy_server(value: &str) -> SystemProxy { + let mut proxy = SystemProxy::default(); + let mut default = None; + + for entry in value + .split(|character: char| character == ';' || character.is_ascii_whitespace()) + .filter(|entry| !entry.is_empty()) + { + let Some((protocol, address)) = entry.split_once('=') else { + default.get_or_insert_with(|| entry.to_string()); + continue; + }; + let address = address.trim(); + if address.is_empty() { + continue; + } + + if protocol.eq_ignore_ascii_case("http") { + proxy.http = Some(address.to_string()); + } else if protocol.eq_ignore_ascii_case("https") { + proxy.https = Some(address.to_string()); + } + } + + proxy.http = proxy.http.or_else(|| default.clone()); + proxy.https = proxy.https.or(default); + proxy +} + +fn normalize_proxy_override(value: &str) -> String { + value + .split(';') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .collect::>() + .join(",") + .replace("*.", "") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn simple_proxy_applies_to_http_and_https() { + let proxy = parse_proxy_server("127.0.0.1:7890"); + + assert_eq!(proxy.http.as_deref(), Some("127.0.0.1:7890")); + assert_eq!(proxy.https.as_deref(), Some("127.0.0.1:7890")); + } + + #[test] + fn protocol_proxy_list_preserves_each_destination_proxy() { + let proxy = parse_proxy_server("http=127.0.0.1:7890;https=secure.example:8443"); + + assert_eq!(proxy.http.as_deref(), Some("127.0.0.1:7890")); + assert_eq!(proxy.https.as_deref(), Some("secure.example:8443")); + } + + #[test] + fn default_proxy_fills_unspecified_protocols() { + let proxy = + parse_proxy_server("HTTP=http.example:8080 fallback.example:3128 ftp=ftp.example:21"); + + assert_eq!(proxy.http.as_deref(), Some("http.example:8080")); + assert_eq!(proxy.https.as_deref(), Some("fallback.example:3128")); + } + + #[test] + fn proxy_override_is_normalized_for_matcher() { + assert_eq!( + normalize_proxy_override("localhost; *.example.com ;10.0.0.0/8"), + "localhost,example.com,10.0.0.0/8" + ); + } +} diff --git a/desktop/src-tauri/Cargo.lock b/desktop/src-tauri/Cargo.lock index 074b8f739e..7018cf565c 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -1023,6 +1023,7 @@ dependencies = [ "buzz-media", "buzz-persona", "buzz-sdk", + "buzz-ws-client", "bytes", "bzip2 0.6.1", "chrono", @@ -1143,6 +1144,23 @@ dependencies = [ "uuid", ] +[[package]] +name = "buzz-ws-client" +version = "0.1.0" +dependencies = [ + "futures-util", + "hyper-util", + "nostr", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-tungstenite 0.29.0", + "tower-service", + "tracing", + "url", + "windows-registry 0.6.1", +] + [[package]] name = "by_address" version = "1.2.1" diff --git a/desktop/src-tauri/Cargo.toml b/desktop/src-tauri/Cargo.toml index d689544688..89da2d7d5e 100644 --- a/desktop/src-tauri/Cargo.toml +++ b/desktop/src-tauri/Cargo.toml @@ -88,6 +88,7 @@ url = "2" buzz_core_pkg = { package = "buzz-core", path = "../../crates/buzz-core" } buzz_persona_pkg = { package = "buzz-persona", path = "../../crates/buzz-persona" } buzz_sdk_pkg = { package = "buzz-sdk", path = "../../crates/buzz-sdk" } +buzz_ws_client_pkg = { package = "buzz-ws-client", path = "../../crates/buzz-ws-client" } buzz_agent_pkg = { package = "buzz-agent", path = "../../crates/buzz-agent" } iroh = { version = "1.0.2", optional = true } mesh-llm-sdk = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "f455d493a2ae82baf2a326e2d0fda351433b4b30", package = "mesh-llm-sdk", default-features = false, features = ["client", "serving"], optional = true } diff --git a/desktop/src-tauri/src/commands/pairing.rs b/desktop/src-tauri/src/commands/pairing.rs index fc874a0150..b1932b39cd 100644 --- a/desktop/src-tauri/src/commands/pairing.rs +++ b/desktop/src-tauri/src/commands/pairing.rs @@ -6,12 +6,13 @@ use buzz_core_pkg::kind::KIND_PAIRING; use buzz_core_pkg::pairing::qr::encode_qr; use buzz_core_pkg::pairing::session::PairingSession; use buzz_core_pkg::pairing::types::{AbortReason, PayloadType}; +use buzz_ws_client_pkg::connect_websocket; use futures_util::{SinkExt, StreamExt}; use nostr::ToBech32; use serde::Serialize; use tauri::{AppHandle, Emitter, State}; use tokio::sync::mpsc; -use tokio_tungstenite::{connect_async, tungstenite::Message}; +use tokio_tungstenite::tungstenite::Message; use tokio_util::sync::CancellationToken; use zeroize::Zeroizing; @@ -264,7 +265,7 @@ async fn pairing_ws_task_inner( outbound_rx: &mut mpsc::Receiver, app: &AppHandle, ) -> Result<(), String> { - let (ws, _) = connect_async(relay_url) + let (ws, _) = connect_websocket(relay_url) .await .map_err(|e| format!("WebSocket connection failed: {e}"))?; let (mut write, mut read) = ws.split(); diff --git a/desktop/src-tauri/src/huddle/relay_api.rs b/desktop/src-tauri/src/huddle/relay_api.rs index eb3fea92d5..db7912d646 100644 --- a/desktop/src-tauri/src/huddle/relay_api.rs +++ b/desktop/src-tauri/src/huddle/relay_api.rs @@ -10,9 +10,10 @@ //! → recv loop: WS binary frame → Opus decode (per-peer) → rodio playback //! ``` +use buzz_ws_client_pkg::connect_websocket; use futures_util::{SinkExt, StreamExt}; use std::sync::{atomic::AtomicBool, Arc}; -use tokio_tungstenite::{connect_async, tungstenite::Message as WsMsg}; +use tokio_tungstenite::tungstenite::Message as WsMsg; use tokio_util::sync::CancellationToken; use uuid::Uuid; @@ -65,7 +66,7 @@ pub(crate) async fn connect_audio_relay( let app_handle = state.app_handle.lock().ok().and_then(|g| g.clone()); - let (ws_stream, _) = connect_async(&ws_url) + let (ws_stream, _) = connect_websocket(&ws_url) .await .map_err(|e| format!("audio WS connect failed: {e}"))?; let (mut ws_tx, mut ws_rx) = ws_stream.split(); diff --git a/desktop/src-tauri/src/native_websocket.rs b/desktop/src-tauri/src/native_websocket.rs index c0cf2e76f1..94df7b7347 100644 --- a/desktop/src-tauri/src/native_websocket.rs +++ b/desktop/src-tauri/src/native_websocket.rs @@ -4,12 +4,11 @@ use futures_util::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; use tauri::{ipc::Channel, plugin::TauriPlugin, Manager, Runtime}; use tokio::sync::{mpsc, oneshot, Mutex}; -use tokio_tungstenite::{ - connect_async, - tungstenite::protocol::{frame::coding::CloseCode, CloseFrame, Message}, -}; +use tokio_tungstenite::tungstenite::protocol::{frame::coding::CloseCode, CloseFrame, Message}; use tokio_util::sync::CancellationToken; +use buzz_ws_client_pkg::connect_websocket; + const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); const WRITE_TIMEOUT: Duration = Duration::from_secs(10); const SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(250); @@ -129,7 +128,7 @@ async fn open_connection( let connect_cancel = manager.connect_cancel.lock().await.clone(); let (socket, _) = tokio::select! { _ = connect_cancel.cancelled() => return Err("WebSocket connection cancelled".to_string()), - result = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(url)) => result + result = tokio::time::timeout(CONNECT_TIMEOUT, connect_websocket(url)) => result .map_err(|_| "WebSocket connection timed out".to_string())? .map_err(|error| error.to_string())?, }; @@ -346,11 +345,9 @@ mod tests { let (_stream, _) = listener.accept().await.unwrap(); tokio::time::sleep(Duration::from_millis(100)).await; }); - let result = std::panic::AssertUnwindSafe(tokio_tungstenite::connect_async(format!( - "wss://{address}" - ))) - .catch_unwind() - .await; + let result = std::panic::AssertUnwindSafe(connect_websocket(format!("wss://{address}"))) + .catch_unwind() + .await; assert!(result.is_ok(), "TLS setup must not panic"); server.await.unwrap();