diff --git a/src-tauri/src/dialer.rs b/src-tauri/src/dialer.rs new file mode 100644 index 00000000..d63676ca --- /dev/null +++ b/src-tauri/src/dialer.rs @@ -0,0 +1,720 @@ +//! Outbound TCP dials of the proxy worker: the destination of a direct route, +//! and the upstream proxy itself. +//! +//! One name lookup per host runs at a time, and a good answer is used again +//! for a short time. The connection then races across the addresses +//! (RFC 8305): the first address starts at once, the next one when an attempt +//! fails or `ATTEMPT_DELAY` passes, and the first connection wins. An address +//! that never answers (often an IPv6 route that drops packets) costs one +//! delay, not the whole time limit. A failure is never remembered: the next +//! dial of the same host starts again. + +use futures_util::future::BoxFuture; +use futures_util::stream::{FuturesUnordered, StreamExt}; +use std::collections::{HashMap, VecDeque}; +use std::future::Future; +use std::io; +use std::net::{IpAddr, SocketAddr}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError}; +use std::time::Duration; +use tokio::net::TcpStream; +use tokio::sync::watch; +use tokio::time::Instant; + +/// Wait before the next address starts while earlier attempts still run. +const ATTEMPT_DELAY: Duration = Duration::from_millis(250); +/// Limit for one attempt while more addresses wait. The last attempt runs +/// until the limit of the dial. +const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(4); +/// Attempts of one dial that run at the same time. Keeps a dial inside the +/// descriptor budget of one connection. +const MAX_IN_FLIGHT: usize = 4; +/// Limit for a caller that waits for a name lookup. +pub(crate) const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5); +/// Until a lookup has run this long, every caller for the same host waits for +/// it. A stuck system resolver then gets one blocking call per host, not one +/// per connection, and each caller still waits for an answer that can come. +const STALE_LOOKUP: Duration = LOOKUP_TIMEOUT; +/// How long a good answer is used again. +const ANSWER_TTL: Duration = Duration::from_secs(30); +const MAX_HOSTS: usize = 2048; + +type Answer = Result, (io::ErrorKind, String)>; +pub(crate) type Lookup = + Arc BoxFuture<'static, io::Result>> + Send + Sync>; + +enum Entry { + Ready { + addrs: Arc<[SocketAddr]>, + until: Instant, + }, + Pending { + id: u64, + started: Instant, + done: watch::Receiver>, + }, +} + +struct ResolverInner { + lookup: Lookup, + hosts: Mutex>, + next_id: AtomicU64, +} + +/// Name lookups shared by every dial of the worker. Cheap to clone. +#[derive(Clone)] +pub(crate) struct HostResolver { + inner: Arc, +} + +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(PoisonError::into_inner) +} + +impl HostResolver { + pub(crate) fn new(lookup: Lookup) -> Self { + Self { + inner: Arc::new(ResolverInner { + lookup, + hosts: Mutex::new(HashMap::new()), + next_id: AtomicU64::new(0), + }), + } + } + + /// The system resolver (`getaddrinfo`). + pub(crate) fn system() -> &'static HostResolver { + static SYSTEM: OnceLock = OnceLock::new(); + SYSTEM.get_or_init(|| { + HostResolver::new(Arc::new( + |host: String| -> BoxFuture<'static, io::Result>> { + Box::pin(async move { Ok(tokio::net::lookup_host((host.as_str(), 0)).await?.collect()) }) + }, + )) + }) + } + + /// Addresses of `host` with port 0, in the order of the resolver. + pub(crate) async fn resolve(&self, host: &str, limit: Duration) -> io::Result> { + let key = host.to_ascii_lowercase(); + let mut done = { + let mut hosts = lock(&self.inner.hosts); + let now = Instant::now(); + match hosts.get(&key) { + Some(Entry::Ready { addrs, until }) if *until > now => return Ok(addrs.clone()), + Some(Entry::Pending { started, done, .. }) + if now.duration_since(*started) < STALE_LOOKUP => + { + done.clone() + } + _ => self.start_lookup(&mut hosts, key), + } + }; + let limit = limit.min(LOOKUP_TIMEOUT); + let answer = match tokio::time::timeout(limit, done.wait_for(Option::is_some)).await { + Ok(Ok(answer)) => (*answer).clone(), + Ok(Err(_)) => None, + Err(_) => { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + format!("name lookup for {host} timed out after {limit:?}"), + )) + } + }; + match answer { + Some(Ok(addrs)) => Ok(addrs), + Some(Err((kind, e))) => Err(io::Error::new( + kind, + format!("name lookup for {host} failed: {e}"), + )), + None => Err(io::Error::other(format!("name lookup for {host} stopped"))), + } + } + + /// The lookup runs in its own task, so it ends and fills the cache even when + /// every caller has stopped waiting. + fn start_lookup( + &self, + hosts: &mut HashMap, + key: String, + ) -> watch::Receiver> { + if hosts.len() >= MAX_HOSTS { + let now = Instant::now(); + hosts.retain(|_, entry| match entry { + Entry::Ready { until, .. } => *until > now, + Entry::Pending { .. } => true, + }); + if hosts.len() >= MAX_HOSTS { + hosts.retain(|_, entry| matches!(entry, Entry::Pending { .. })); + } + } + let (tx, rx) = watch::channel(None); + let id = self.inner.next_id.fetch_add(1, Ordering::Relaxed); + hosts.insert( + key.clone(), + Entry::Pending { + id, + started: Instant::now(), + done: rx.clone(), + }, + ); + let lookup = (self.inner.lookup)(key.clone()); + let inner = Arc::clone(&self.inner); + tokio::spawn(async move { + let answer: Answer = match lookup.await { + Ok(found) => { + let mut addrs: Vec = Vec::with_capacity(found.len()); + for addr in found { + if !addrs.contains(&addr) { + addrs.push(addr); + } + } + if addrs.is_empty() { + Err((io::ErrorKind::NotFound, "no address".to_string())) + } else { + Ok(addrs.into()) + } + } + Err(e) => Err((e.kind(), e.to_string())), + }; + inner.finish(&key, id, &answer); + let _ = tx.send(Some(answer)); + }); + rx + } + + /// Put `winner` first for the next dials of `host`, so an address that + /// does not answer costs the delay once per answer, not once per dial. + pub(crate) fn prefer(&self, host: &str, winner: IpAddr) { + let mut hosts = lock(&self.inner.hosts); + let Some(Entry::Ready { addrs, .. }) = hosts.get_mut(&host.to_ascii_lowercase()) else { + return; + }; + if let Some(position) = addrs.iter().position(|a| a.ip() == winner) { + if position > 0 { + let mut reordered = addrs.to_vec(); + let first = reordered.remove(position); + reordered.insert(0, first); + *addrs = reordered.into(); + } + } + } +} + +impl ResolverInner { + /// A good answer is kept; a failed one is dropped, so the next caller looks + /// the name up again. + fn finish(&self, key: &str, id: u64, answer: &Answer) { + let mut hosts = lock(&self.hosts); + match answer { + Ok(addrs) => { + hosts.insert( + key.to_string(), + Entry::Ready { + addrs: addrs.clone(), + until: Instant::now() + ANSWER_TTL, + }, + ); + } + Err(_) => { + if matches!(hosts.get(key), Some(Entry::Pending { id: pending, .. }) if *pending == id) { + hosts.remove(key); + } + } + } + } +} + +/// Address families of the attempts of one dial. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) struct Families { + ipv4: bool, + ipv6: bool, +} + +impl Families { + fn add(&mut self, addr: SocketAddr) { + if addr.is_ipv4() { + self.ipv4 = true; + } else { + self.ipv6 = true; + } + } +} + +impl std::fmt::Display for Families { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match (self.ipv4, self.ipv6) { + (true, true) => "ipv4,ipv6", + (true, false) => "ipv4", + (false, true) => "ipv6", + (false, false) => "none", + }) + } +} + +/// Timings of a race across addresses. +pub(crate) struct Race { + /// Wait before the next address starts while earlier attempts still run. + pub stagger: Duration, + /// Limit for one attempt while more addresses wait. + pub attempt_timeout: Duration, + /// The dial fails at this instant. + pub deadline: Instant, +} + +/// Why no attempt connected. +#[derive(Debug)] +pub(crate) struct RaceFailure { + pub attempts: usize, + pub families: Families, + pub deadline_hit: bool, + pub last: Option, +} + +impl RaceFailure { + /// One error for the log. `what` names the dial: `direct connect to h:443`. + /// It never names an address: a log line must not list visited sites. + pub(crate) fn into_error(self, what: &str, limit: Duration) -> io::Error { + let tail = format!("attempts={} families={}", self.attempts, self.families); + match self.last { + _ if self.deadline_hit => io::Error::new( + io::ErrorKind::TimedOut, + format!("{what} timed out after {limit:?} {tail}"), + ), + Some(last) => io::Error::new(last.kind(), format!("{what} failed: {last} {tail}")), + None => io::Error::new( + io::ErrorKind::InvalidInput, + format!("{what} failed: no address {tail}"), + ), + } + } +} + +/// RFC 8305 order: the families alternate, starting with the family of the +/// first address. +fn interleave(addrs: Vec) -> Vec { + let Some(first) = addrs.first() else { + return addrs; + }; + let lead_is_v6 = first.is_ipv6(); + let (lead, other): (Vec<_>, Vec<_>) = addrs + .into_iter() + .partition(|addr| addr.is_ipv6() == lead_is_v6); + let mut order = Vec::with_capacity(lead.len() + other.len()); + let mut lead = lead.into_iter(); + let mut other = other.into_iter(); + loop { + match (lead.next(), other.next()) { + (None, None) => return order, + (a, b) => order.extend(a.into_iter().chain(b)), + } + } +} + +/// Run `connect` across `addrs` and return the first connection with its +/// address. The losing attempts are dropped. +pub(crate) async fn race( + addrs: Vec, + plan: &Race, + mut connect: F, +) -> Result<(T, SocketAddr), RaceFailure> +where + F: FnMut(SocketAddr) -> Fut, + Fut: Future>, +{ + let mut waiting: VecDeque = interleave(addrs).into(); + let mut running = FuturesUnordered::new(); + let mut failure = RaceFailure { + attempts: 0, + families: Families::default(), + deadline_hit: false, + last: None, + }; + let mut next_start = Instant::now(); + loop { + let now = Instant::now(); + if running.len() < MAX_IN_FLIGHT && (running.is_empty() || now >= next_start) { + if let Some(addr) = waiting.pop_front() { + failure.attempts += 1; + failure.families.add(addr); + let limit = if waiting.is_empty() { + plan.deadline + } else { + (now + plan.attempt_timeout).min(plan.deadline) + }; + let attempt = connect(addr); + running.push(async move { (addr, tokio::time::timeout_at(limit, attempt).await) }); + next_start = now + plan.stagger; + continue; + } + } + if running.is_empty() { + failure.deadline_hit = Instant::now() >= plan.deadline; + return Err(failure); + } + let may_start = running.len() < MAX_IN_FLIGHT && !waiting.is_empty(); + tokio::select! { + Some((addr, outcome)) = running.next() => { + let error = match outcome { + Ok(Ok(value)) => return Ok((value, addr)), + Ok(Err(e)) => e, + Err(_) => io::Error::new(io::ErrorKind::TimedOut, "attempt timed out"), + }; + log::debug!("Dial attempt failed addr={addr} err=\"{error}\""); + failure.last = Some(error); + next_start = Instant::now(); + } + () = tokio::time::sleep_until(next_start), if may_start => {} + () = tokio::time::sleep_until(plan.deadline) => { + failure.deadline_hit = true; + return Err(failure); + } + } + } +} + +/// Race TCP connections to `addrs` with the timings of RFC 8305. +pub(crate) async fn connect_addrs( + what: &str, + addrs: Vec, + deadline: Instant, + limit: Duration, +) -> io::Result<(TcpStream, SocketAddr)> { + let plan = Race { + stagger: ATTEMPT_DELAY, + attempt_timeout: ATTEMPT_TIMEOUT, + deadline, + }; + match race(addrs, &plan, TcpStream::connect).await { + Ok((stream, addr)) => { + let _ = stream.set_nodelay(true); + Ok((stream, addr)) + } + Err(failure) => Err(failure.into_error(what, limit)), + } +} + +/// Connect to `host:port` within `limit`, name lookup included. `role` starts +/// the error text: `direct`, `upstream proxy`. +pub(crate) async fn connect_host( + role: &str, + host: &str, + port: u16, + limit: Duration, +) -> io::Result { + connect_host_with(HostResolver::system(), role, host, port, limit).await +} + +async fn connect_host_with( + resolver: &HostResolver, + role: &str, + host: &str, + port: u16, + limit: Duration, +) -> io::Result { + let deadline = Instant::now() + limit; + let host = host + .strip_prefix('[') + .and_then(|h| h.strip_suffix(']')) + .unwrap_or(host); + let what = format!( + "{role} connect to {}", + crate::proxy_server::connect_authority(host, port) + ); + if let Ok(ip) = host.parse::() { + let addrs = vec![SocketAddr::new(ip, port)]; + return Ok(connect_addrs(&what, addrs, deadline, limit).await?.0); + } + let addrs = resolver + .resolve(host, limit) + .await? + .iter() + .map(|addr| { + let mut addr = *addr; + addr.set_port(port); + addr + }) + .collect(); + let (stream, winner) = connect_addrs(&what, addrs, deadline, limit).await?; + resolver.prefer(host, winner.ip()); + Ok(stream) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::AtomicUsize; + + fn plan(attempt_timeout: Duration, limit: Duration) -> Race { + Race { + stagger: ATTEMPT_DELAY, + attempt_timeout, + deadline: Instant::now() + limit, + } + } + + /// A port on this machine where nothing listens. + async fn closed_port() -> SocketAddr { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + listener.local_addr().unwrap() + } + + fn resolver_of( + addrs: Vec, + calls: Arc, + fail_first: bool, + ) -> HostResolver { + HostResolver::new(Arc::new( + move |_host: String| -> BoxFuture<'static, io::Result>> { + let call = calls.fetch_add(1, Ordering::SeqCst); + let addrs = addrs.clone(); + Box::pin(async move { + tokio::time::sleep(Duration::from_millis(100)).await; + if fail_first && call == 0 { + Err(io::Error::other("resolver not ready")) + } else { + Ok(addrs) + } + }) + }, + )) + } + + #[test] + fn families_alternate_from_the_first_answer() { + let v6 = |n: u16| { + SocketAddr::new( + IpAddr::V6(std::net::Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, n)), + 443, + ) + }; + let v4 = |n: u8| SocketAddr::new(IpAddr::V4(std::net::Ipv4Addr::new(192, 0, 2, n)), 443); + assert_eq!( + interleave(vec![v6(1), v6(2), v6(3), v4(1), v4(2)]), + vec![v6(1), v4(1), v6(2), v4(2), v6(3)] + ); + assert_eq!( + interleave(vec![v4(1), v4(2), v6(1)]), + vec![v4(1), v6(1), v4(2)] + ); + assert!(interleave(Vec::new()).is_empty()); + } + + #[tokio::test] + async fn an_address_that_never_answers_costs_one_delay() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let good = listener.local_addr().unwrap(); + let black_hole: SocketAddr = "192.0.2.1:9".parse().unwrap(); + let started = std::time::Instant::now(); + let (_, winner) = race( + vec![black_hole, good], + &plan(Duration::from_secs(4), Duration::from_secs(10)), + |addr| async move { + if addr == black_hole { + std::future::pending::<()>().await; + } + TcpStream::connect(addr).await + }, + ) + .await + .expect("the second address connects"); + assert_eq!(winner, good); + assert!( + started.elapsed() < Duration::from_secs(1), + "{:?}", + started.elapsed() + ); + } + + #[tokio::test] + async fn a_real_unreachable_first_address_does_not_use_the_whole_limit() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let good = listener.local_addr().unwrap(); + let started = std::time::Instant::now(); + let (_, winner) = connect_addrs( + "direct connect to test", + vec!["10.255.255.1:9".parse().unwrap(), good], + Instant::now() + Duration::from_secs(10), + Duration::from_secs(10), + ) + .await + .expect("the local listener connects"); + assert_eq!(winner, good); + assert!( + started.elapsed() < Duration::from_secs(2), + "{:?}", + started.elapsed() + ); + } + + #[tokio::test] + async fn when_every_address_fails_the_error_counts_the_attempts() { + let addrs = vec![closed_port().await, closed_port().await]; + let error = connect_addrs( + "direct connect to test", + addrs, + Instant::now() + Duration::from_secs(5), + Duration::from_secs(5), + ) + .await + .expect_err("nothing listens"); + let text = error.to_string(); + assert!(text.contains("attempts=2"), "{text}"); + assert!(text.contains("families=ipv4"), "{text}"); + assert_ne!(error.kind(), io::ErrorKind::TimedOut, "{text}"); + + let silent: Vec = vec![ + "192.0.2.1:9".parse().unwrap(), + "[2001:db8::1]:9".parse().unwrap(), + ]; + let failure = race( + silent, + &plan(Duration::from_millis(200), Duration::from_millis(600)), + |_| std::future::pending::>(), + ) + .await + .expect_err("nothing answers"); + assert_eq!(failure.attempts, 2); + assert!(failure.deadline_hit); + let error = failure.into_error("direct connect to test", Duration::from_millis(600)); + assert_eq!(error.kind(), io::ErrorKind::TimedOut); + assert!(error.to_string().contains("families=ipv4,ipv6"), "{error}"); + } + + #[tokio::test] + async fn concurrent_dials_of_one_host_share_one_lookup() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let calls = Arc::new(AtomicUsize::new(0)); + let resolver = resolver_of(vec!["127.0.0.1:0".parse().unwrap()], calls.clone(), false); + let dials = (0..16).map(|_| { + let resolver = resolver.clone(); + tokio::spawn(async move { + connect_host_with( + &resolver, + "direct", + "Example.Test", + port, + Duration::from_secs(5), + ) + .await + }) + }); + for dial in futures_util::future::join_all(dials).await { + dial.unwrap().expect("dial through the shared lookup"); + } + assert_eq!(calls.load(Ordering::SeqCst), 1); + connect_host_with( + &resolver, + "direct", + "example.test", + port, + Duration::from_secs(5), + ) + .await + .expect("cached answer"); + assert_eq!(calls.load(Ordering::SeqCst), 1, "the answer is reused"); + } + + #[tokio::test] + async fn a_failed_dial_does_not_fail_the_next_dial_of_the_host() { + let calls = Arc::new(AtomicUsize::new(0)); + let resolver = resolver_of(vec!["127.0.0.1:0".parse().unwrap()], calls.clone(), true); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + connect_host_with( + &resolver, + "direct", + "example.test", + port, + Duration::from_secs(5), + ) + .await + .expect_err("the first lookup fails"); + connect_host_with( + &resolver, + "direct", + "example.test", + port, + Duration::from_secs(5), + ) + .await + .expect("a failed lookup is not remembered"); + assert_eq!(calls.load(Ordering::SeqCst), 2); + + let refused = closed_port().await.port(); + let error = connect_host_with( + &resolver, + "direct", + "example.test", + refused, + Duration::from_secs(5), + ) + .await + .expect_err("nothing listens"); + assert_eq!(error.kind(), io::ErrorKind::ConnectionRefused, "{error}"); + let started = std::time::Instant::now(); + connect_host_with( + &resolver, + "direct", + "example.test", + port, + Duration::from_secs(5), + ) + .await + .expect("a failed connect is not remembered"); + assert!(started.elapsed() < Duration::from_secs(1)); + } + + #[tokio::test] + async fn the_address_that_answered_goes_first_next_time() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let calls = Arc::new(AtomicUsize::new(0)); + let resolver = resolver_of( + vec![ + "127.0.0.2:0".parse().unwrap(), + "127.0.0.1:0".parse().unwrap(), + ], + calls, + false, + ); + // 127.0.0.2 refuses on Linux and times out on macOS; either way the + // second address wins. + connect_host_with( + &resolver, + "direct", + "example.test", + port, + Duration::from_secs(5), + ) + .await + .expect("the second address connects"); + let order = resolver + .resolve("example.test", LOOKUP_TIMEOUT) + .await + .unwrap(); + assert_eq!(order[0].ip(), "127.0.0.1".parse::().unwrap()); + } + + #[tokio::test] + async fn ip_literals_skip_the_lookup() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let calls = Arc::new(AtomicUsize::new(0)); + let resolver = resolver_of(Vec::new(), calls.clone(), false); + connect_host_with( + &resolver, + "direct", + "127.0.0.1", + port, + Duration::from_secs(5), + ) + .await + .expect("literal"); + assert_eq!(calls.load(Ordering::SeqCst), 0); + } +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 13e3c497..717f6905 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -133,6 +133,7 @@ mod browser_version_manager; mod cdp_target; mod checksum; mod default_browser; +mod dialer; pub mod dns_blocklist; mod downloaded_browsers_registry; mod downloader; diff --git a/src-tauri/src/proxy_server.rs b/src-tauri/src/proxy_server.rs index d4e64fe2..9570c0fc 100644 --- a/src-tauri/src/proxy_server.rs +++ b/src-tauri/src/proxy_server.rs @@ -469,15 +469,20 @@ impl AsyncWrite for SocksHandshakeLogger { } async fn connect_via_socks( - socks_addr: &str, + socks_host: &str, + socks_port: u16, target_host: &str, target_port: u16, is_socks5: bool, auth: Option<(&str, &str)>, ) -> Result> { - let stream = tokio::time::timeout(UPSTREAM_DIAL_TIMEOUT, TcpStream::connect(socks_addr)) - .await - .map_err(|_| format!("SOCKS upstream connect to {socks_addr} timed out"))??; + let stream = crate::dialer::connect_host( + "SOCKS upstream", + socks_host, + socks_port, + UPSTREAM_DIAL_TIMEOUT, + ) + .await?; if is_socks5 { // SOCKS5 connection using async_socks5 @@ -496,8 +501,9 @@ async fn connect_via_socks( let has_auth = auth_info.is_some(); log::trace!( - "[socks-handshake] dialing {} (target={}:{}, has_auth={})", - socks_addr, + "[socks-handshake] dialing {}:{} (target={}:{}, has_auth={})", + socks_host, + socks_port, target_host, target_port, has_auth @@ -514,7 +520,7 @@ async fn connect_via_socks( // and silently FIN instead of returning an RFC1929 status. BufStream // coalesces those small writes into one syscall on flush — this is // the usage pattern shown in the async_socks5 README - let label = format!("{socks_addr}->{target_host}:{target_port}"); + let label = format!("{socks_host}:{socks_port}->{target_host}:{target_port}"); let logged = SocksHandshakeLogger::new(stream, label); let mut buffered = tokio::io::BufStream::new(logged); let handshake = tokio::time::timeout( @@ -878,7 +884,6 @@ async fn handle_http_via_socks4( let socks_host = upstream.host_str().unwrap_or("127.0.0.1"); let socks_port = upstream.port().unwrap_or(1080); - let socks_addr = format!("{}:{}", socks_host, socks_port); // Parse target from request URI let target_uri = req.uri(); @@ -886,26 +891,28 @@ async fn handle_http_via_socks4( let target_port = target_uri.port_u16().unwrap_or(80); // Connect to SOCKS4 proxy - let mut socks_stream = - match tokio::time::timeout(UPSTREAM_DIAL_TIMEOUT, TcpStream::connect(&socks_addr)).await { - Ok(Ok(stream)) => stream, - Ok(Err(e)) => { - warn_upstream_failure("SOCKS4 upstream connect failed", &e.to_string(), None); - let mut response = Response::new(Full::new(Bytes::from(format!( - "Failed to connect to SOCKS4 proxy: {}", - e - )))); - *response.status_mut() = StatusCode::BAD_GATEWAY; - return Ok(response); - } - Err(_) => { - warn_upstream_failure("SOCKS4 upstream connect failed", "timed out", None); - let mut response = - Response::new(Full::new(Bytes::from("Connect to SOCKS4 proxy timed out"))); - *response.status_mut() = StatusCode::GATEWAY_TIMEOUT; - return Ok(response); - } - }; + let mut socks_stream = match crate::dialer::connect_host( + "SOCKS4 upstream", + socks_host, + socks_port, + UPSTREAM_DIAL_TIMEOUT, + ) + .await + { + Ok(stream) => stream, + Err(e) => { + warn_upstream_failure("SOCKS4 upstream connect failed", &e.to_string(), None); + let mut response = Response::new(Full::new(Bytes::from(format!( + "Failed to connect to SOCKS4 proxy: {e}" + )))); + *response.status_mut() = if e.kind() == io::ErrorKind::TimedOut { + StatusCode::GATEWAY_TIMEOUT + } else { + StatusCode::BAD_GATEWAY + }; + return Ok(response); + } + }; // Build a SOCKS4a CONNECT request. We deliberately do NOT resolve the target // hostname locally: tokio::net::lookup_host would call the HOST resolver @@ -1172,6 +1179,29 @@ async fn forward_http_over_stream( Ok(hyper_response) } +/// Dial the Shadowsocks server and open a tunnel to `target` over it. The +/// tunnel header goes out with the first write. +async fn connect_via_shadowsocks( + ss_host: &str, + ss_port: u16, + svr_cfg: &shadowsocks::config::ServerConfig, + target: shadowsocks::relay::Address, +) -> io::Result> { + let tcp = crate::dialer::connect_host( + "Shadowsocks server", + ss_host, + ss_port, + UPSTREAM_DIAL_TIMEOUT, + ) + .await?; + let context = shadowsocks::context::Context::new_shared(shadowsocks::config::ServerType::Local); + Ok( + shadowsocks::relay::tcprelay::proxy_stream::ProxyClientStream::from_stream( + context, tcp, svr_cfg, target, + ), + ) +} + /// Handle plain HTTP requests through a Shadowsocks upstream. /// reqwest doesn't support SS natively, so we connect through the SS tunnel /// manually and forward the HTTP request/response. @@ -1206,7 +1236,6 @@ async fn handle_http_via_shadowsocks( } }; - let context = shadowsocks::context::Context::new_shared(shadowsocks::config::ServerType::Local); let svr_cfg = match shadowsocks::config::ServerConfig::new( shadowsocks::config::ServerAddr::from((ss_host.to_string(), ss_port)), &password, @@ -1222,13 +1251,7 @@ async fn handle_http_via_shadowsocks( let target_addr = shadowsocks::relay::Address::DomainNameAddress(domain.clone(), port); - let mut stream = match shadowsocks::relay::tcprelay::proxy_stream::ProxyClientStream::connect( - context, - &svr_cfg, - target_addr, - ) - .await - { + let mut stream = match connect_via_shadowsocks(ss_host, ss_port, &svr_cfg, target_addr).await { Ok(s) => s, Err(e) => { let mut resp = Response::new(Full::new(Bytes::from(format!("SS connect: {e}")))); @@ -2097,14 +2120,7 @@ async fn handle_connect_from_buffer( return Err("Invalid CONNECT request".into()); } - let target = parts[1]; - let (target_host, target_port) = if let Some(colon_pos) = target.find(':') { - let host = &target[..colon_pos]; - let port: u16 = target[colon_pos + 1..].parse().unwrap_or(443); - (host, port) - } else { - (target, 443) - }; + let (target_host, target_port) = split_authority(parts[1]); // Block if domain is in the DNS blocklist (before any connection) if blocklist_matcher.is_blocked(target_host) { @@ -2217,11 +2233,17 @@ fn worker_connection_limit() -> usize { } } -/// Connect timeout for the direct (no-upstream) dial path. Bounds a wedged -/// `getaddrinfo` so a broken resolver can't park a blocking thread for the -/// full OS timeout. +/// Limit for a direct (no-upstream) dial, name lookup included. const DIRECT_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +async fn dial_direct(host: &str, port: u16) -> io::Result { + crate::dialer::connect_host("direct", host, port, DIRECT_CONNECT_TIMEOUT).await +} + +/// With a profile DNS server and an upstream, the wait before the second +/// address of a name gets its own dial through the upstream. +const UPSTREAM_ATTEMPT_DELAY: std::time::Duration = std::time::Duration::from_secs(2); + /// Overall timeout for dialing an UPSTREAM proxy (TCP connect + CONNECT/SOCKS/SS /// handshake). Without it, an upstream that accepts TCP but stalls before /// replying hangs the worker task forever and holds a connection slot; under @@ -2249,91 +2271,6 @@ const MAX_HTTP_RESPONSE_BUFFER: usize = 10 * 1024 * 1024; /// the body and break all three. const PLAIN_HTTP_EXCHANGE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60); -/// Per-host failure state (last failure instant, consecutive failure count) for -/// the direct dial path. Process-global — each worker is its own process. -fn direct_dial_failures() -> &'static Mutex> { - static M: OnceLock>> = OnceLock::new(); - M.get_or_init(|| Mutex::new(HashMap::new())) -} - -/// If `host` is inside its failure backoff window, return the remaining time so -/// the caller can short-circuit without a fresh getaddrinfo/connect. Never -/// mutates state, so the window always expires and the path self-heals once -/// DNS recovers. -fn direct_backoff_remaining(host: &str) -> Option { - let map = direct_dial_failures(); - let guard = map.lock().unwrap(); - let (last, fails) = guard.get(host).copied()?; - // Exponential window capped at 30s: 2, 4, 8, 16, 30, 30, ... - let window = std::time::Duration::from_secs((1u64 << fails.min(5)).min(30)); - let elapsed = last.elapsed(); - if elapsed < window { - Some(window - elapsed) - } else { - None - } -} - -/// Record a direct-dial failure for `host`, growing its backoff window. -fn direct_backoff_record(host: &str) { - let map = direct_dial_failures(); - let mut guard = map.lock().unwrap(); - // Bound memory against a page that emits many distinct failing hosts. - if guard.len() > 2048 { - guard.retain(|_, (last, _)| last.elapsed() < std::time::Duration::from_secs(60)); - } - let entry = guard - .entry(host.to_string()) - .or_insert_with(|| (std::time::Instant::now(), 0)); - entry.0 = std::time::Instant::now(); - entry.1 = entry.1.saturating_add(1); -} - -/// Clear `host`'s failure state after a successful dial. -fn direct_backoff_clear(host: &str) { - direct_dial_failures().lock().unwrap().remove(host); -} - -/// Dial a target directly (no upstream) with a connect timeout and per-host -/// failure backoff. This is the server-side counterpart to the browser's -/// instant client-side retry: when a host's DNS/connect is failing (e.g. the -/// macOS resolver wedges after sleep/wake), repeated CONNECT requests -/// short-circuit -/// here instead of each spawning a fresh blocking getaddrinfo — which is what -/// let a retry-storm exhaust the blocking thread pool and peg the CPU. -async fn dial_direct(host: &str, port: u16) -> Result> { - if let Some(remaining) = direct_backoff_remaining(host) { - return Err( - format!( - "skipping direct dial to {host}: backing off ~{}s after repeated connect failures", - remaining.as_secs().max(1) - ) - .into(), - ); - } - match tokio::time::timeout(DIRECT_CONNECT_TIMEOUT, TcpStream::connect((host, port))).await { - Ok(Ok(stream)) => { - let _ = stream.set_nodelay(true); - direct_backoff_clear(host); - Ok(stream) - } - Ok(Err(e)) => { - direct_backoff_record(host); - Err(e.into()) - } - Err(_) => { - direct_backoff_record(host); - Err( - format!( - "direct connect to {host}:{port} timed out after {}s", - DIRECT_CONNECT_TIMEOUT.as_secs() - ) - .into(), - ) - } - } -} - /// Host of a buffered `CONNECT host:port HTTP/1.1` request. fn connect_target_host(request: &[u8]) -> Option { let line_end = request @@ -2341,14 +2278,23 @@ fn connect_target_host(request: &[u8]) -> Option { .position(|&b| b == b'\r' || b == b'\n') .unwrap_or(request.len()); let line = String::from_utf8_lossy(&request[..line_end]); - let target = line.split_whitespace().nth(1)?; - let host = match target.strip_prefix('[') { - Some(rest) => rest.split(']').next()?, - None => target.rsplit_once(':').map_or(target, |(host, _)| host), - }; + let (host, _) = split_authority(line.split_whitespace().nth(1)?); (!host.is_empty()).then(|| host.to_string()) } +/// Host and port of a CONNECT authority, `host:port` or `[v6]:port`. The port +/// defaults to 443. +fn split_authority(target: &str) -> (&str, u16) { + if let Some((host, rest)) = target.strip_prefix('[').and_then(|t| t.split_once(']')) { + let port = rest.strip_prefix(':').and_then(|p| p.parse().ok()); + return (host, port.unwrap_or(443)); + } + match target.split_once(':') { + Some((host, port)) => (host, port.parse().unwrap_or(443)), + None => (target, 443), + } +} + /// Warn about a failed tunnel or proxied request, at most once per distinct /// message every 30s. The target host is cut out of the message: these lines /// reach the file, and the file must not list the sites a profile visited. @@ -2414,6 +2360,17 @@ async fn read_upstream_connect_response( } } +/// `host:port` for a CONNECT line. An IPv6 literal (a SOCKS5 client sends one +/// for `http://[2001:db8::1]/`) needs brackets, or the proxy reads its last +/// group as the port. +pub(crate) fn connect_authority(host: &str, port: u16) -> String { + if host.parse::().is_ok() { + format!("[{host}]:{port}") + } else { + format!("{host}:{port}") + } +} + /// Perform the HTTP CONNECT handshake over an already-established hop to the /// proxy and return the tunnelled stream. /// @@ -2422,17 +2379,6 @@ async fn read_upstream_connect_response( /// only place `Proxy-Authorization` is written, so whether those credentials /// cross the network in the clear is decided entirely by which stream the /// caller hands in, nothing here can weaken it. -/// `host:port` for a CONNECT line. An IPv6 literal (a SOCKS5 client sends one -/// for `http://[2001:db8::1]/`) needs brackets, or the proxy reads its last -/// group as the port. -fn connect_authority(host: &str, port: u16) -> String { - if host.parse::().is_ok() { - format!("[{host}]:{port}") - } else { - format!("{host}:{port}") - } -} - async fn connect_via_http_proxy( mut proxy_stream: S, proxy_host: &str, @@ -2544,16 +2490,50 @@ pub(crate) async fn connect_to_target_via_upstream( let Some(resolver) = resolver.filter(|_| !should_bypass) else { return dial_route(target_host, target_port, upstream_url, should_bypass).await; }; - let addresses = resolver.resolve(target_host).await?; - // A second address covers an exit that cannot reach the first one. - let mut last_error = String::new(); - for address in addresses.iter().take(2) { - match dial_route(&address.to_string(), target_port, upstream_url, false).await { - Ok(stream) => return Ok(stream), - Err(e) => last_error = e.to_string(), - } + let addresses: Vec = resolver + .resolve(target_host) + .await? + .iter() + .map(|ip| SocketAddr::new(*ip, target_port)) + .collect(); + let authority = connect_authority(target_host, target_port); + if matches!(upstream_url, None | Some("DIRECT")) { + let (stream, _) = crate::dialer::connect_addrs( + &format!("direct connect to {authority}"), + addresses, + tokio::time::Instant::now() + DIRECT_CONNECT_TIMEOUT, + DIRECT_CONNECT_TIMEOUT, + ) + .await?; + return Ok(Box::new(stream)); + } + // Each attempt is a whole dial through the upstream, so the next address + // waits longer, and only a second address gets a turn: it covers an exit + // that cannot reach the first one. + let limit = 2 * UPSTREAM_DIAL_TIMEOUT; + let plan = crate::dialer::Race { + stagger: UPSTREAM_ATTEMPT_DELAY, + attempt_timeout: UPSTREAM_DIAL_TIMEOUT, + deadline: tokio::time::Instant::now() + limit, + }; + let candidates = addresses.into_iter().take(2).collect(); + let raced = crate::dialer::race(candidates, &plan, |address| async move { + dial_route(&address.ip().to_string(), target_port, upstream_url, false) + .await + .map_err(|e| io::Error::other(e.to_string())) + }) + .await; + match raced { + Ok((stream, _)) => Ok(stream), + Err(failure) => Err( + failure + .into_error( + &format!("connect to {authority} through the upstream"), + limit, + ) + .into(), + ), } - Err(last_error.into()) } /// Dial `target_host:target_port` over the route as given, with no name @@ -2564,10 +2544,6 @@ pub(crate) async fn dial_route( upstream_url: Option<&str>, should_bypass: bool, ) -> Result> { - // Helper: configure outbound TCP to match browser TCP fingerprint - let configure_tcp = |stream: &TcpStream| { - let _ = stream.set_nodelay(true); - }; let target_stream: BoxedAsyncStream = match upstream_url { None | Some("DIRECT") => Box::new(dial_direct(target_host, target_port).await?), _ if should_bypass => Box::new(dial_direct(target_host, target_port).await?), @@ -2585,15 +2561,13 @@ pub(crate) async fn dial_route( "http" | "https" => { let proxy_host = upstream.host_str().unwrap_or("127.0.0.1"); let proxy_port = upstream.port().unwrap_or(8080); - let proxy_stream = tokio::time::timeout( + let proxy_stream = crate::dialer::connect_host( + "upstream proxy", + proxy_host, + proxy_port, UPSTREAM_DIAL_TIMEOUT, - TcpStream::connect((proxy_host, proxy_port)), ) - .await - .map_err(|_| { - format!("upstream proxy connect to {proxy_host}:{proxy_port} timed out") - })??; - configure_tcp(&proxy_stream); + .await?; connect_via_http_proxy( proxy_stream, @@ -2611,15 +2585,13 @@ pub(crate) async fn dial_route( "httpstls" => { let proxy_host = upstream.host_str().unwrap_or("127.0.0.1"); let proxy_port = upstream.port().unwrap_or(443); - let tcp = tokio::time::timeout( + let tcp = crate::dialer::connect_host( + "upstream proxy", + proxy_host, + proxy_port, UPSTREAM_DIAL_TIMEOUT, - TcpStream::connect((proxy_host, proxy_port)), ) - .await - .map_err(|_| { - format!("upstream proxy connect to {proxy_host}:{proxy_port} timed out") - })??; - configure_tcp(&tcp); + .await?; let tls = tls_wrap_upstream_hop(tcp, proxy_host).await?; @@ -2636,13 +2608,13 @@ pub(crate) async fn dial_route( "socks4" | "socks5" => { let socks_host = upstream.host_str().unwrap_or("127.0.0.1"); let socks_port = upstream.port().unwrap_or(1080); - let socks_addr = format!("{}:{}", socks_host, socks_port); let (username, password) = upstream_userpass(&upstream); let auth = (!username.is_empty()).then_some((username.as_str(), password.as_str())); let stream = connect_via_socks( - &socks_addr, + socks_host, + socks_port, target_host, target_port, scheme == "socks5", @@ -2677,8 +2649,6 @@ pub(crate) async fn dial_route( format!("Unsupported Shadowsocks cipher: {method_str}. Use e.g. aes-256-gcm, chacha20-ietf-poly1305, aes-128-gcm") })?; - let context = - shadowsocks::context::Context::new_shared(shadowsocks::config::ServerType::Local); let svr_cfg = shadowsocks::config::ServerConfig::new( shadowsocks::config::ServerAddr::from((ss_host.to_string(), ss_port)), &password, @@ -2693,19 +2663,7 @@ pub(crate) async fn dial_route( } }; - let stream = tokio::time::timeout( - UPSTREAM_DIAL_TIMEOUT, - shadowsocks::relay::tcprelay::proxy_stream::ProxyClientStream::connect( - context, - &svr_cfg, - target_addr, - ), - ) - .await - .map_err(|_| "Shadowsocks connection timed out".to_string())? - .map_err(|e| format!("Shadowsocks connection failed: {e}"))?; - - Box::new(stream) + Box::new(connect_via_shadowsocks(ss_host, ss_port, &svr_cfg, target_addr).await?) } _ => { return Err(format!("Unsupported upstream proxy scheme: {}", scheme).into()); @@ -3026,7 +2984,7 @@ mod tests { use tokio::io::{AsyncReadExt, AsyncWriteExt}; let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let address = listener.local_addr().unwrap().to_string(); + let port = listener.local_addr().unwrap().port(); let server = tokio::spawn(async move { let (mut socket, _) = listener.accept().await.unwrap(); let mut request = vec![0_u8; 8 + 1 + "example.com".len() + 1]; @@ -3038,7 +2996,7 @@ mod tests { request }); - connect_via_socks(&address, "example.com", 443, false, None) + connect_via_socks("127.0.0.1", port, "example.com", 443, false, None) .await .expect("SOCKS4a connect"); let request = server.await.unwrap(); @@ -3053,6 +3011,18 @@ mod tests { assert_eq!(connect_authority("example.com", 443), "example.com:443"); } + #[test] + fn a_connect_target_with_an_ipv6_literal_keeps_its_address() { + assert_eq!(split_authority("[2001:db8::1]:8443"), ("2001:db8::1", 8443)); + assert_eq!(split_authority("[2001:db8::1]"), ("2001:db8::1", 443)); + assert_eq!(split_authority("example.com:80"), ("example.com", 80)); + assert_eq!(split_authority("example.com"), ("example.com", 443)); + assert_eq!( + connect_target_host(b"CONNECT [2001:db8::1]:443 HTTP/1.1\r\n\r\n").as_deref(), + Some("2001:db8::1") + ); + } + #[test] fn a_fully_qualified_name_with_a_trailing_dot_is_still_blocked() { let matcher = matcher_of(&["tracker.com"]); diff --git a/src-tauri/src/socks5_local.rs b/src-tauri/src/socks5_local.rs index 86d7c5ef..9ae42a64 100644 --- a/src-tauri/src/socks5_local.rs +++ b/src-tauri/src/socks5_local.rs @@ -573,12 +573,13 @@ async fn associate_upstream( None }; - let proxy_stream = tokio::time::timeout( + let proxy_stream = crate::dialer::connect_host( + "upstream SOCKS5", + host, + port, crate::proxy_server::UPSTREAM_DIAL_TIMEOUT, - TcpStream::connect((host, port)), ) - .await - .map_err(|_| format!("upstream SOCKS5 connect to {host}:{port} timed out"))??; + .await?; let bind_sock = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).await?; // association_addr None => 0.0.0.0:0 (we accept replies from any peer). let datagram = tokio::time::timeout( @@ -792,7 +793,8 @@ async fn run_udp_relay_socks5( } /// Resolve a UDP destination to a concrete socket address, with the profile -/// DNS server when there is one and the system resolver otherwise. +/// DNS server when there is one and the system resolver otherwise. The direct +/// egress socket is IPv4, so an IPv4 answer goes first. async fn resolve_addr(addr: &AddrKind, resolver: Option<&ProfileResolver>) -> Option { match (addr, resolver) { (AddrKind::Ip(s), _) => Some(*s), @@ -802,10 +804,18 @@ async fn resolve_addr(addr: &AddrKind, resolver: Option<&ProfileResolver>) -> Op .ok()? .first() .map(|ip| SocketAddr::new(*ip, *port)), - (AddrKind::Domain(domain, port), None) => tokio::net::lookup_host(format!("{domain}:{port}")) - .await - .ok() - .and_then(|mut it| it.next()), + (AddrKind::Domain(domain, port), None) => { + let found = crate::dialer::HostResolver::system() + .resolve(domain, crate::dialer::LOOKUP_TIMEOUT) + .await + .ok()?; + let ip = found + .iter() + .find(|a| a.is_ipv4()) + .or_else(|| found.first())? + .ip(); + Some(SocketAddr::new(ip, *port)) + } } }