diff --git a/Cargo.lock b/Cargo.lock index da86c89b851..1391ce1453e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -805,6 +805,7 @@ dependencies = [ "buzz-core", "buzz-persona", "buzz-sdk", + "buzz-ws-client", "chrono", "clap", "evalexpr", @@ -1102,6 +1103,7 @@ name = "buzz-pairing-cli" version = "0.1.0" dependencies = [ "buzz-core", + "buzz-ws-client", "clap", "futures-util", "hex", @@ -1372,13 +1374,18 @@ name = "buzz-ws-client" version = "0.1.0" dependencies = [ "futures-util", + "hyper-util", "nostr", + "plist", "serde_json", + "system-configuration", "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 98d68fb5215..3ef9a4a515e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -119,8 +119,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 d047849806f..f616c79fc70 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 e6164b02dd3..2c0190b2a9e 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 2cbb82411fd..1769f2ace9c 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -120,12 +120,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; @@ -452,7 +453,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), @@ -3362,17 +3363,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`. @@ -3668,7 +3679,7 @@ pub(crate) fn parse_relay_message(text: &str) -> Result bool { match err { @@ -3724,7 +3735,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, } } @@ -3847,7 +3858,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)))?; @@ -4364,7 +4375,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")) @@ -6029,7 +6040,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"), ))); @@ -6037,6 +6048,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 58fefe32d96..4f432fbb83d 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 1eb9d215f92..553421c0047 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 5cec925677f..680c7fc36ce 100644 --- a/crates/buzz-ws-client/Cargo.toml +++ b/crates/buzz-ws-client/Cargo.toml @@ -10,8 +10,17 @@ 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(target_os = "macos")'.dependencies] +plist = "1" +system-configuration = "0.7" + +[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 bec5b56bb43..15e750fe703 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 02c7d4b1b20..23a64409cd6 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/macos.rs b/crates/buzz-ws-client/src/proxy/macos.rs new file mode 100644 index 00000000000..ebd7743b2df --- /dev/null +++ b/crates/buzz-ws-client/src/proxy/macos.rs @@ -0,0 +1,218 @@ +//! macOS static proxy discovery. +//! +//! Hyper-util reads the configured HTTP and HTTPS proxies on macOS, but does +//! not currently include `ExceptionsList` or `ExcludeSimpleHostnames`. This +//! module preserves its environment-variable precedence while adding those +//! system bypass settings to Buzz's shared matcher. + +use std::io::Cursor; + +use hyper_util::client::proxy::matcher::Matcher; +use plist::{Dictionary, Value}; +use system_configuration::{ + core_foundation::{ + base::{CFType, TCFType}, + dictionary::CFDictionary, + propertylist::{create_data, kCFPropertyListBinaryFormat_v1_0}, + string::CFString, + }, + dynamic_store::SCDynamicStoreBuilder, +}; + +use super::SystemProxySettings; + +pub(super) fn system_proxy_settings() -> SystemProxySettings { + if std::env::var_os("REQUEST_METHOD").is_some() { + return SystemProxySettings::new(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"]); + let mut exclude_simple_hostnames = false; + + 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; + exclude_simple_hostnames = system.exclude_simple_hostnames; + } + } + + SystemProxySettings { + matcher: Matcher::builder() + .all(all) + .http(http) + .https(https) + .no(no) + .build(), + exclude_simple_hostnames, + } +} + +fn first_env(names: &[&str]) -> String { + names + .iter() + .find_map(|name| std::env::var(name).ok()) + .unwrap_or_default() +} + +fn read_system_proxy() -> Option { + let store = SCDynamicStoreBuilder::new("buzz-ws-client").build()?; + let proxies = store.get_proxies()?; + let dictionary = proxy_dictionary(&proxies)?; + Some(parse_system_proxy(&dictionary)) +} + +fn proxy_dictionary(proxies: &CFDictionary) -> Option { + let data = create_data( + proxies.as_CFTypeRef().cast(), + kCFPropertyListBinaryFormat_v1_0, + ) + .ok()?; + Value::from_reader(Cursor::new(data.bytes())) + .ok()? + .into_dictionary() +} + +#[derive(Debug, Default, PartialEq, Eq)] +struct SystemProxy { + http: Option, + https: Option, + no: String, + exclude_simple_hostnames: bool, +} + +fn parse_system_proxy(proxies: &Dictionary) -> SystemProxy { + SystemProxy { + http: proxy_endpoint(proxies, "HTTPEnable", "HTTPProxy", "HTTPPort"), + https: proxy_endpoint(proxies, "HTTPSEnable", "HTTPSProxy", "HTTPSPort"), + no: proxy_exceptions(proxies), + exclude_simple_hostnames: setting_enabled(proxies.get("ExcludeSimpleHostnames")), + } +} + +fn proxy_endpoint( + proxies: &Dictionary, + enabled_key: &str, + host_key: &str, + port_key: &str, +) -> Option { + if !setting_enabled(proxies.get(enabled_key)) { + return None; + } + + let host = proxies.get(host_key)?.as_string()?.trim(); + if host.is_empty() { + return None; + } + + let port = proxies + .get(port_key) + .and_then(integer_value) + .and_then(|value| u16::try_from(value).ok()); + Some(match port { + Some(port) => format!("{host}:{port}"), + None => host.to_string(), + }) +} + +fn proxy_exceptions(proxies: &Dictionary) -> String { + proxies + .get("ExceptionsList") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(Value::as_string) + .filter_map(normalize_exception) + .collect::>() + .join(",") +} + +fn normalize_exception(value: &str) -> Option<&str> { + let value = value.trim(); + let value = value.strip_prefix("*.").unwrap_or(value); + (!value.is_empty()).then_some(value) +} + +fn setting_enabled(value: Option<&Value>) -> bool { + value.is_some_and(|value| { + value.as_boolean() == Some(true) + || value.as_signed_integer() == Some(1) + || value.as_unsigned_integer() == Some(1) + }) +} + +fn integer_value(value: &Value) -> Option { + value.as_signed_integer().or_else(|| { + value + .as_unsigned_integer() + .and_then(|value| i64::try_from(value).ok()) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_enabled_http_and_https_proxies() { + let mut proxies = Dictionary::new(); + proxies.insert("HTTPEnable".into(), Value::Integer(1.into())); + proxies.insert("HTTPProxy".into(), Value::String("proxy.local".into())); + proxies.insert("HTTPPort".into(), Value::Integer(8080.into())); + proxies.insert("HTTPSEnable".into(), Value::Boolean(true)); + proxies.insert( + "HTTPSProxy".into(), + Value::String("secure-proxy.local".into()), + ); + proxies.insert("HTTPSPort".into(), Value::Integer(8443.into())); + + let proxy = parse_system_proxy(&proxies); + + assert_eq!(proxy.http.as_deref(), Some("proxy.local:8080")); + assert_eq!(proxy.https.as_deref(), Some("secure-proxy.local:8443")); + } + + #[test] + fn normalizes_macos_proxy_exceptions_for_hyper_util() { + let mut proxies = Dictionary::new(); + proxies.insert( + "ExceptionsList".into(), + Value::Array(vec![ + Value::String(" *.example.com ".into()), + Value::String("10.0.0.0/8".into()), + Value::String("localhost".into()), + Value::String(String::new()), + ]), + ); + + let exceptions = proxy_exceptions(&proxies); + assert_eq!(exceptions, "example.com,10.0.0.0/8,localhost"); + + let matcher = Matcher::builder() + .all("http://proxy.local:8080") + .no(exceptions) + .build(); + let excluded = "https://relay.example.com".parse().unwrap(); + let proxied = "https://notexample.com".parse().unwrap(); + assert!(matcher.intercept(&excluded).is_none()); + assert!(matcher.intercept(&proxied).is_some()); + } + + #[test] + fn reads_exclude_simple_hostnames_boolean_and_integer_values() { + let mut proxies = Dictionary::new(); + proxies.insert("ExcludeSimpleHostnames".into(), Value::Integer(1.into())); + assert!(parse_system_proxy(&proxies).exclude_simple_hostnames); + + proxies.insert("ExcludeSimpleHostnames".into(), Value::Boolean(false)); + assert!(!parse_system_proxy(&proxies).exclude_simple_hostnames); + } +} 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 00000000000..3e6c9b58b60 --- /dev/null +++ b/crates/buzz-ws-client/src/proxy/mod.rs @@ -0,0 +1,488 @@ +//! 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(target_os = "macos")] +mod macos; +#[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 settings = system_proxy_settings(); + connect_websocket_with_options( + &settings.matcher, + request, + settings.exclude_simple_hostnames, + ) + .await +} + +struct SystemProxySettings { + matcher: Matcher, + exclude_simple_hostnames: bool, +} + +impl SystemProxySettings { + fn new(matcher: Matcher) -> Self { + Self { + matcher, + exclude_simple_hostnames: false, + } + } +} + +fn system_proxy_settings() -> SystemProxySettings { + #[cfg(target_os = "macos")] + { + macos::system_proxy_settings() + } + + #[cfg(windows)] + { + SystemProxySettings::new(windows::system_proxy_matcher()) + } + + #[cfg(not(any(target_os = "macos", windows)))] + { + SystemProxySettings::new(Matcher::from_system()) + } +} + +#[cfg(test)] +async fn connect_websocket_with( + matcher: &Matcher, + request: Request, +) -> Result<(ProxyWebSocketStream, Response), Error> { + connect_websocket_with_options(matcher, request, false).await +} + +async fn connect_websocket_with_options( + matcher: &Matcher, + request: Request, + exclude_simple_hostnames: bool, +) -> Result<(ProxyWebSocketStream, Response), Error> { + let target = proxy_target_uri(request.uri())?; + let stream = connect_tcp(matcher, &target, exclude_simple_hostnames).await?; + + client_async_tls(request, stream).await +} + +async fn connect_tcp( + matcher: &Matcher, + target: &Uri, + exclude_simple_hostnames: bool, +) -> Result { + if target.host().is_some_and(|host| { + is_loopback_host(host) || (exclude_simple_hostnames && is_simple_hostname(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 is_simple_hostname(host: &str) -> bool { + let host = host.trim_matches(['[', ']']); + !host.contains('.') && host.parse::().is_err() +} + +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")); + } + + #[test] + fn simple_hostnames_exclude_ip_addresses() { + assert!(is_simple_hostname("relay")); + assert!(!is_simple_hostname("relay.example.com")); + assert!(!is_simple_hostname("192.0.2.1")); + assert!(!is_simple_hostname("[2001:db8::1]")); + } + + #[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 00000000000..b1a9264a686 --- /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 23caff0ae54..53e4c22908a 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -1078,6 +1078,7 @@ dependencies = [ "buzz-sdk", "buzz-terminal", "buzz-voice", + "buzz-ws-client", "bytes", "bzip2 0.6.1", "chrono", @@ -1234,6 +1235,25 @@ dependencies = [ "tokenizers", ] +[[package]] +name = "buzz-ws-client" +version = "0.1.0" +dependencies = [ + "futures-util", + "hyper-util", + "nostr", + "plist", + "serde_json", + "system-configuration", + "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 463bf88dd89..86e84c4cd50 100644 --- a/desktop/src-tauri/Cargo.toml +++ b/desktop/src-tauri/Cargo.toml @@ -105,6 +105,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" } buzz_voice_pkg = { package = "buzz-voice", path = "../../crates/buzz-voice" } buzz_terminal = { package = "buzz-terminal", path = "crates/buzz-terminal" } diff --git a/desktop/src-tauri/src/commands/pairing.rs b/desktop/src-tauri/src/commands/pairing.rs index aedd67854c1..4c0e27a06ba 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, Manager, 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; @@ -303,7 +304,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 3f2aa76a560..f773a3bd17f 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 128f2df79dd..2b385e5711c 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())?, }; @@ -358,11 +357,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();