From c09ee5cc534c8893680fc33814022ae7f22bcb7f Mon Sep 17 00:00:00 2001 From: Gregory Schier Date: Fri, 6 Nov 2026 12:52:30 -0700 Subject: [PATCH] Skip Private Relay for the proxy engine's plain-HTTP upstream on macOS (#818) --- Cargo.lock | 4 + crates/yaak-proxy/Cargo.toml | 6 + crates/yaak-proxy/src/lib.rs | 26 +- crates/yaak-proxy/src/nw.rs | 1078 +++++++++++++++++++++++ crates/yaak-proxy/src/request.rs | 59 +- crates/yaak-proxy/src/tunnel.rs | 85 +- crates/yaak-proxy/tests/own_fallback.rs | 118 +++ crates/yaak-proxy/tests/proxy.rs | 36 +- 8 files changed, 1377 insertions(+), 35 deletions(-) create mode 100644 crates/yaak-proxy/src/nw.rs create mode 100644 crates/yaak-proxy/tests/own_fallback.rs diff --git a/Cargo.lock b/Cargo.lock index 005fe1520..927a8f5b8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11878,13 +11878,16 @@ name = "yaak-proxy" version = "0.1.0" dependencies = [ "base64 0.23.1", + "block2 0.6.1", "bytes", + "dispatch2", "flate2", "futures-util", "http", "http-body-util", "hyper", "hyper-util", + "libc", "log 0.4.34", "pem", "rcgen", @@ -11896,6 +11899,7 @@ dependencies = [ "tokio", "tokio-rustls", "tokio-tungstenite", + "tower-service", ] [[package]] diff --git a/crates/yaak-proxy/Cargo.toml b/crates/yaak-proxy/Cargo.toml index 736231482..efeaf1178 100644 --- a/crates/yaak-proxy/Cargo.toml +++ b/crates/yaak-proxy/Cargo.toml @@ -28,6 +28,12 @@ tokio-rustls = "0.26" pem = "3" time = "0.3" +[target.'cfg(target_os = "macos")'.dependencies] +block2 = "0.6" +dispatch2 = "0.3" +libc = "0.2" +tower-service = "0.3" + [dev-dependencies] futures-util = "0.3" hyper-util = { version = "0.1", features = ["server-auto"] } diff --git a/crates/yaak-proxy/src/lib.rs b/crates/yaak-proxy/src/lib.rs index 81f70685d..81f962535 100644 --- a/crates/yaak-proxy/src/lib.rs +++ b/crates/yaak-proxy/src/lib.rs @@ -4,6 +4,8 @@ mod connection; mod decode; mod grpc; mod messages; +#[cfg(target_os = "macos")] +mod nw; mod request; mod sse; mod tunnel; @@ -21,6 +23,7 @@ use std::time::SystemTime; use bytes::Bytes; use http::HeaderValue; use hyper_util::client::legacy::Client; +#[cfg(not(target_os = "macos"))] use hyper_util::client::legacy::connect::HttpConnector; use hyper_util::rt::TokioExecutor; use rustls::ClientConfig; @@ -242,7 +245,7 @@ pub(crate) struct Engine { events: std_mpsc::Sender, next_id: AtomicU64, next_connection: AtomicU64, - http_client: Client, + http_client: HttpClient, upstream_tls: Arc, upstream_tls_h1: Arc, passthrough: Mutex>, @@ -250,8 +253,8 @@ pub(crate) struct Engine { /// Offer HTTP/2 to clients in intercepted TLS. http2: AtomicBool, websocket_capture: bool, - /// `Via` value marking requests whose URL the engine rebuilt from `Host`, unique to this run, - /// so one that comes back to the engine is refused instead of forwarded again. + /// `Via` value marking the plain-HTTP requests the engine forwards, unique to this run, so one + /// that comes back to the engine is refused instead of forwarded again. via: HeaderValue, } @@ -305,6 +308,21 @@ fn upstream_tls_config( Ok(Arc::new(config)) } +/// Sends plain HTTP upstream. On macOS it connects through Network.framework, so Private Relay +/// can't route the engine's own port-80 connections back into the engine (see [`nw`]). +#[cfg(target_os = "macos")] +type HttpClient = Client; +#[cfg(not(target_os = "macos"))] +type HttpClient = Client; + +fn http_client() -> HttpClient { + let builder = Client::builder(TokioExecutor::new()); + #[cfg(target_os = "macos")] + return builder.build(nw::Connector); + #[cfg(not(target_os = "macos"))] + builder.build_http() +} + fn via_marker() -> HeaderValue { let run = RandomState::new().build_hasher().finish(); HeaderValue::try_from(format!("1.1 yaak-proxy-{run:016x}")).unwrap() @@ -327,7 +345,7 @@ pub fn start_proxy(config: ProxyConfig) -> Result { events: event_tx, next_id: AtomicU64::new(1), next_connection: AtomicU64::new(1), - http_client: Client::builder(TokioExecutor::new()).build_http(), + http_client: http_client(), upstream_tls: upstream_tls_config(&roots, &[b"h2", b"http/1.1"])?, upstream_tls_h1: upstream_tls_config(&roots, &[b"http/1.1"])?, passthrough: Mutex::new(HashSet::new()), diff --git a/crates/yaak-proxy/src/nw.rs b/crates/yaak-proxy/src/nw.rs new file mode 100644 index 000000000..a5f901cbf --- /dev/null +++ b/crates/yaak-proxy/src/nw.rs @@ -0,0 +1,1078 @@ +//! Upstream TCP through Network.framework, for macOS. +//! +//! While a system web proxy is set, iCloud Private Relay hands every public port-80 connection +//! to that proxy, raw sockets included. When the proxy is this engine, its own upstream for +//! plain HTTP comes straight back to it. A Network.framework connection with `prefer_no_proxy` +//! goes to the origin instead. "Prefer" means it falls back to the proxy when there is no +//! direct route, so [`connect`] refuses a connection made through a proxy on this Mac, and the +//! `Via` loop guard stays in place behind it. + +use std::collections::HashMap; +use std::ffi::{CStr, CString, c_char, c_void}; +use std::future::Future; +use std::io; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV6}; +use std::pin::Pin; +use std::ptr; +use std::sync::{Arc, LazyLock, Mutex, OnceLock}; +use std::task::{Context, Poll, Waker}; +use std::time::{Duration, Instant}; + +use block2::{Block, RcBlock}; +use bytes::Bytes; +use dispatch2::{DispatchData, DispatchQueue, DispatchRetained}; +use hyper::Uri; +use hyper::http::uri::Scheme; +use hyper_util::client::legacy::connect::{Connected, Connection as HyperConnection}; +use hyper_util::rt::TokioIo; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +use tokio::sync::{oneshot, watch}; + +type NwObject = *mut c_void; + +const STATE_WAITING: u32 = 1; +const STATE_READY: u32 = 3; +const STATE_FAILED: u32 = 4; +const STATE_CANCELLED: u32 = 5; +const ENDPOINT_TYPE_ADDRESS: u32 = 1; +const ENDPOINT_TYPE_HOST: u32 = 2; +const ENDPOINT_TYPE_URL: u32 = 4; +const ERROR_DOMAIN_POSIX: u32 = 1; +const ERROR_DOMAIN_DNS: u32 = 2; + +/// Largest chunk asked of one receive. +const MAX_READ: u32 = 64 * 1024; +/// Bytes handed to Network.framework but not yet taken by the stack. Writes wait above this. +const MAX_IN_FLIGHT: usize = 256 * 1024; + +#[link(name = "Network", kind = "framework")] +unsafe extern "C" { + static _nw_parameters_configure_protocol_disable: NwObject; + static _nw_parameters_configure_protocol_default_configuration: NwObject; + static _nw_content_context_default_message: NwObject; + static _nw_content_context_final_send: NwObject; + + fn nw_release(obj: NwObject); + fn nw_endpoint_create_host(host: *const c_char, port: *const c_char) -> NwObject; + fn nw_endpoint_get_type(endpoint: NwObject) -> u32; + fn nw_endpoint_get_address(endpoint: NwObject) -> *const libc::sockaddr; + fn nw_endpoint_get_hostname(endpoint: NwObject) -> *const c_char; + fn nw_endpoint_get_url(endpoint: NwObject) -> *const c_char; + fn nw_parameters_create_secure_tcp( + configure_tls: NwObject, + configure_tcp: NwObject, + ) -> NwObject; + fn nw_parameters_set_prefer_no_proxy(parameters: NwObject, prefer_no_proxy: bool); + fn nw_connection_create(endpoint: NwObject, parameters: NwObject) -> NwObject; + fn nw_connection_set_queue(connection: NwObject, queue: NwObject); + fn nw_connection_set_state_changed_handler( + connection: NwObject, + handler: &Block, + ); + fn nw_connection_start(connection: NwObject); + fn nw_connection_cancel(connection: NwObject); + fn nw_connection_copy_current_path(connection: NwObject) -> NwObject; + fn nw_path_copy_effective_remote_endpoint(path: NwObject) -> NwObject; + fn nw_connection_send( + connection: NwObject, + content: NwObject, + context: NwObject, + is_complete: bool, + completion: &Block, + ); + fn nw_connection_receive( + connection: NwObject, + minimum_incomplete_length: u32, + maximum_length: u32, + // `is_complete` is a C `bool`, which block2 can't encode; `u8` has the same ABI. + completion: &Block, + ); + fn nw_content_context_get_is_final(context: NwObject) -> bool; + fn nw_connection_access_establishment_report( + connection: NwObject, + queue: NwObject, + access_block: &Block, + ); + fn nw_establishment_report_get_used_proxy(report: NwObject) -> bool; + fn nw_establishment_report_copy_proxy_endpoint(report: NwObject) -> NwObject; + fn nw_error_get_error_domain(error: NwObject) -> u32; + fn nw_error_get_error_code(error: NwObject) -> i32; +} + +/// An `nw_error_t`, kept as its domain and code so it can be reported more than once. +#[derive(Clone, Copy, Debug)] +struct NwError { + domain: u32, + code: i32, +} + +impl NwError { + /// Reads `error`, an `nw_error_t`; `None` when it is null. + fn from_raw(error: NwObject) -> Option { + if error.is_null() { + return None; + } + // SAFETY: `error` is a live `nw_error_t` for the duration of the callback. + unsafe { + Some(Self { + domain: nw_error_get_error_domain(error), + code: nw_error_get_error_code(error), + }) + } + } + + fn to_io(self) -> io::Error { + match self.domain { + ERROR_DOMAIN_POSIX => io::Error::from_raw_os_error(self.code), + ERROR_DOMAIN_DNS => { + io::Error::other(format!("failed to look up address (DNS error {})", self.code)) + } + _ => io::Error::other(format!( + "Network.framework error (domain {}, code {})", + self.domain, self.code + )), + } + } +} + +/// An `nw_connection_t`, cancelled when dropped. +struct Connection(NwObject); + +// SAFETY: Network.framework objects are thread-safe. +unsafe impl Send for Connection {} +unsafe impl Sync for Connection {} + +impl Drop for Connection { + fn drop(&mut self) { + // SAFETY: `self.0` is a connection this wrapper owns a reference to. + unsafe { + nw_connection_cancel(self.0); + nw_release(self.0); + } + } +} + +impl Connection { + /// The address the connection reached, from its current path. + fn remote_addr(&self) -> Option { + // SAFETY: the path and endpoint are copies released here, and the address they hand out + // lives as long as the endpoint. + unsafe { + let path = nw_connection_copy_current_path(self.0); + if path.is_null() { + return None; + } + let endpoint = nw_path_copy_effective_remote_endpoint(path); + nw_release(path); + if endpoint.is_null() { + return None; + } + let addr = if nw_endpoint_get_type(endpoint) == ENDPOINT_TYPE_ADDRESS { + socket_addr(nw_endpoint_get_address(endpoint)) + } else { + None + }; + nw_release(endpoint); + addr + } + } + + /// Whether the ready connection went through a proxy on this Mac instead of direct. + async fn used_local_proxy(&self) -> bool { + let (tx, rx) = oneshot::channel(); + { + let tx = Mutex::new(Some(tx)); + let on_report = RcBlock::new(move |report: NwObject| { + // SAFETY: a non-null `report` is live for the callback, and the endpoint copied + // from it is released here. + let local = !report.is_null() + && unsafe { + nw_establishment_report_get_used_proxy(report) && { + let endpoint = nw_establishment_report_copy_proxy_endpoint(report); + let local = endpoint.is_null() || is_local_endpoint(endpoint); + if !endpoint.is_null() { + nw_release(endpoint); + } + local + } + }; + if let Some(tx) = tx.lock().unwrap().take() { + let _ = tx.send(local); + } + }); + let queue = DispatchQueue::new("app.yaak.proxy.upstream-report", None); + // SAFETY: the connection is live; the queue is retained and the block copied. + unsafe { + let queue = DispatchRetained::as_ptr(&queue).as_ptr().cast(); + nw_connection_access_establishment_report(self.0, queue, &on_report); + } + } + // The report comes exactly once; without it, assume a direct connection. + rx.await.unwrap_or(false) + } + + /// Sends `data`, or with `None` closes the sending side (a FIN). + fn send(&self, data: Option<&[u8]>, done: &Block) { + let content = data.map(DispatchData::from_bytes); + let content_ptr = + content.as_ref().map_or(ptr::null_mut(), |d| DispatchRetained::as_ptr(d).as_ptr()); + // SAFETY: the connection and content are live; Network.framework retains the content and + // copies the block. + unsafe { + let context = if data.is_some() { + _nw_content_context_default_message + } else { + _nw_content_context_final_send + }; + nw_connection_send(self.0, content_ptr.cast(), context, true, done); + } + } +} + +/// # Safety +/// `sa` is null or points to a valid `sockaddr` of the length its family implies. +unsafe fn socket_addr(sa: *const libc::sockaddr) -> Option { + if sa.is_null() { + return None; + } + // SAFETY: the family says which `sockaddr_*` `sa` points to. + unsafe { + match i32::from((*sa).sa_family) { + libc::AF_INET => { + let sin = &*sa.cast::(); + let ip = Ipv4Addr::from(u32::from_be(sin.sin_addr.s_addr)); + Some(SocketAddr::from((ip, u16::from_be(sin.sin_port)))) + } + libc::AF_INET6 => { + let sin6 = &*sa.cast::(); + let ip = Ipv6Addr::from(sin6.sin6_addr.s6_addr); + let port = u16::from_be(sin6.sin6_port); + Some(SocketAddrV6::new(ip, port, sin6.sin6_flowinfo, sin6.sin6_scope_id).into()) + } + _ => None, + } + } +} + +/// What one receive delivered. +struct Received { + data: Bytes, + /// The peer closed its sending side after `data`. + eof: bool, + /// The connection failed after `data`. + error: Option, +} + +#[derive(Default)] +struct ReadState { + /// A receive was issued and has not completed. + pending: bool, + received: Option, + waker: Option, +} + +#[derive(Default)] +struct WriteState { + /// Sends issued whose completion has not run, counting the FIN. + sends: usize, + bytes: usize, + /// The first send that failed. Every later write, flush and shutdown reports it. + error: Option, + waker: Option, +} + +/// State the completion handlers share with the stream; they run on the connection's queue. +#[derive(Default)] +struct Shared { + read: Mutex, + write: Mutex, +} + +/// A TCP connection opened by Network.framework with `prefer_no_proxy`. Reads ask for one chunk +/// at a time, so a slow reader closes the TCP window; writes wait once [`MAX_IN_FLIGHT`] bytes +/// are queued. Shutdown sends a FIN, and dropping the stream cancels the connection. +pub(crate) struct NwStream { + conn: Connection, + shared: Arc, + remote_addr: Option, + buffered: Bytes, + eof: bool, + read_error: Option, + fin_sent: bool, +} + +impl NwStream { + pub(crate) fn remote_addr(&self) -> Option { + self.remote_addr + } + + fn receive(&self) { + let shared = self.shared.clone(); + let done = RcBlock::new( + move |content: NwObject, context: NwObject, is_complete: u8, error: NwObject| { + let data = if content.is_null() { + Bytes::new() + } else { + // SAFETY: a non-null `content` is a `dispatch_data_t` live for the callback. + Bytes::from(unsafe { &*content.cast::() }.to_vec()) + }; + // SAFETY: a non-null `context` is a live `nw_content_context_t`. + let eof = is_complete != 0 + && (context.is_null() || unsafe { nw_content_context_get_is_final(context) }); + let received = Received { data, eof, error: NwError::from_raw(error) }; + let mut read = shared.read.lock().unwrap(); + read.pending = false; + read.received = Some(received); + if let Some(waker) = read.waker.take() { + waker.wake(); + } + }, + ); + // SAFETY: the connection is live and Network.framework copies the block. + unsafe { nw_connection_receive(self.conn.0, 1, MAX_READ, &done) }; + } + + fn send(&self, data: Option<&[u8]>) { + let shared = self.shared.clone(); + let len = data.map_or(0, <[u8]>::len); + { + let mut write = shared.write.lock().unwrap(); + write.sends += 1; + write.bytes += len; + } + let done = RcBlock::new(move |error: NwObject| { + let mut write = shared.write.lock().unwrap(); + write.sends -= 1; + write.bytes -= len; + if write.error.is_none() { + write.error = NwError::from_raw(error); + } + if let Some(waker) = write.waker.take() { + waker.wake(); + } + }); + self.conn.send(data, &done); + } +} + +impl AsyncRead for NwStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.get_mut(); + if buf.remaining() == 0 { + return Poll::Ready(Ok(())); + } + loop { + if !this.buffered.is_empty() { + let n = this.buffered.len().min(buf.remaining()); + buf.put_slice(&this.buffered.split_to(n)); + return Poll::Ready(Ok(())); + } + if let Some(error) = this.read_error.take() { + return Poll::Ready(Err(error.to_io())); + } + if this.eof { + return Poll::Ready(Ok(())); + } + + let mut read = this.shared.read.lock().unwrap(); + if let Some(received) = read.received.take() { + drop(read); + this.buffered = received.data; + this.eof = received.eof; + this.read_error = received.error; + continue; + } + read.waker = Some(cx.waker().clone()); + if !read.pending { + read.pending = true; + drop(read); + this.receive(); + } + return Poll::Pending; + } + } +} + +impl AsyncWrite for NwStream { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let this = self.get_mut(); + if this.fin_sent { + return Poll::Ready(Err(io::ErrorKind::BrokenPipe.into())); + } + { + let mut write = this.shared.write.lock().unwrap(); + if let Some(error) = write.error { + return Poll::Ready(Err(error.to_io())); + } + if write.bytes >= MAX_IN_FLIGHT { + write.waker = Some(cx.waker().clone()); + return Poll::Pending; + } + } + if buf.is_empty() { + return Poll::Ready(Ok(0)); + } + this.send(Some(buf)); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let mut write = self.shared.write.lock().unwrap(); + if write.sends > 0 { + write.waker = Some(cx.waker().clone()); + return Poll::Pending; + } + Poll::Ready(write.error.map_or(Ok(()), |e| Err(e.to_io()))) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + if !self.fin_sent { + std::task::ready!(self.as_mut().poll_flush(cx))?; + self.fin_sent = true; + self.send(None); + } + self.poll_flush(cx) + } +} + +/// How long an open that went through a proxy on this Mac marks its target as the engine's own. +const OWN_FALLBACK_WINDOW: Duration = Duration::from_secs(10); +/// Longest wait in [`is_own_fallback`] for the opens of the same target to finish, all together. +/// The engine's own fallback finishes within milliseconds of the engine answering its `CONNECT`. +const OWN_FALLBACK_WAIT: Duration = Duration::from_secs(2); + +/// One open by [`connect`] that is under way, or recently went through a proxy on this Mac. +struct Attempt { + /// `None` while connecting, then whether it went through a proxy on this Mac. + outcome: watch::Sender>, + ended: OnceLock, +} + +/// Every [`Attempt`] by normalized `host:port`. +static ATTEMPTS: LazyLock>>>> = + LazyLock::new(Mutex::default); + +fn attempt_key(host: &str, port: u16) -> String { + let host = host.trim_start_matches('[').trim_end_matches(']').trim_end_matches('.'); + format!("{}:{port}", host.to_ascii_lowercase()) +} + +/// Drops the attempts that ended without a local proxy, and those past the window. +fn prune_attempts(attempts: &mut HashMap>>) { + attempts.retain(|_, list| { + list.retain(|a| match a.ended.get() { + None => true, + Some(ended) => { + *a.outcome.borrow() == Some(true) && ended.elapsed() < OWN_FALLBACK_WINDOW + } + }); + !list.is_empty() + }); +} + +/// Registers an open for [`is_own_fallback`]; ends it as direct when dropped unfinished. +struct AttemptGuard(Arc); + +impl AttemptGuard { + fn start(host: &str, port: u16) -> Self { + let attempt = + Arc::new(Attempt { outcome: watch::Sender::new(None), ended: OnceLock::new() }); + let mut attempts = ATTEMPTS.lock().unwrap(); + prune_attempts(&mut attempts); + attempts.entry(attempt_key(host, port)).or_default().push(attempt.clone()); + Self(attempt) + } + + /// Records the outcome; only the first call counts. + fn finish(&self, via_local_proxy: bool) { + self.0.outcome.send_if_modified(|outcome| { + outcome.get_or_insert(via_local_proxy); + true + }); + let _ = self.0.ended.set(Instant::now()); + } +} + +impl Drop for AttemptGuard { + fn drop(&mut self) { + self.finish(false); + } +} + +/// Whether a `CONNECT` to `host:port` may be one of the engine's own connections that fell back +/// to the system proxy, which is this engine: an open of the same target went through a proxy on +/// this Mac. Waits for opens of it that are under way, since the one that sent the `CONNECT` +/// finishes as soon as the engine answers it. An open that went direct never sent one. They are +/// watched together, up to [`OWN_FALLBACK_WAIT`] in all: the first through a local proxy answers +/// yes, and once every one has finished without, no. +pub(crate) async fn is_own_fallback(host: &str, port: u16) -> bool { + let pending = { + let mut attempts = ATTEMPTS.lock().unwrap(); + prune_attempts(&mut attempts); + attempts.get(&attempt_key(host, port)).cloned().unwrap_or_default() + }; + let mut outcomes = tokio::task::JoinSet::new(); + for attempt in pending { + let mut outcome = attempt.outcome.subscribe(); + outcomes.spawn(async move { + outcome.wait_for(Option::is_some).await.is_ok_and(|o| *o == Some(true)) + }); + } + let deadline = tokio::time::Instant::now() + OWN_FALLBACK_WAIT; + // Dropping the set at the end aborts the waits still running. + while let Ok(Some(via_local_proxy)) = + tokio::time::timeout_at(deadline, outcomes.join_next()).await + { + if via_local_proxy.unwrap_or(false) { + return true; + } + } + false +} + +/// Opens a TCP connection to `host:port` that skips the system proxy where it can. +pub(crate) async fn connect(host: &str, port: u16) -> io::Result { + open(host, port, |_| {}).await +} + +/// [`connect`], with `configure` given the `nw_parameters_t` before the connection is created. +async fn open(host: &str, port: u16, configure: impl FnOnce(NwObject)) -> io::Result { + let invalid = || io::Error::new(io::ErrorKind::InvalidInput, format!("invalid host {host:?}")); + let c_host = CString::new(host).map_err(|_| invalid())?; + let c_port = CString::new(port.to_string()).map_err(|_| invalid())?; + + // SAFETY: the C strings outlive the call, and each object created here is released once. + let conn = unsafe { + let endpoint = nw_endpoint_create_host(c_host.as_ptr(), c_port.as_ptr()); + if endpoint.is_null() { + return Err(invalid()); + } + let parameters = nw_parameters_create_secure_tcp( + _nw_parameters_configure_protocol_disable, + _nw_parameters_configure_protocol_default_configuration, + ); + nw_parameters_set_prefer_no_proxy(parameters, true); + configure(parameters); + let conn = nw_connection_create(endpoint, parameters); + nw_release(endpoint); + nw_release(parameters); + if conn.is_null() { + return Err(invalid()); + } + Connection(conn) + }; + + let attempt = AttemptGuard::start(host, port); + + // A refused or unroutable connection goes to `waiting` to retry when the network changes; + // fail it there instead, as a socket would. + let (ready_tx, ready_rx) = oneshot::channel::>(); + { + let ready_tx = Mutex::new(Some(ready_tx)); + let on_state = RcBlock::new(move |state: u32, error: NwObject| { + let result = match (state, NwError::from_raw(error)) { + (STATE_READY, _) => Ok(()), + (STATE_WAITING | STATE_FAILED, Some(error)) => Err(error), + (STATE_FAILED | STATE_CANCELLED, None) => { + Err(NwError { domain: ERROR_DOMAIN_POSIX, code: libc::ECONNABORTED }) + } + _ => return, + }; + if let Some(tx) = ready_tx.lock().unwrap().take() { + let _ = tx.send(result); + } + }); + let queue = DispatchQueue::new("app.yaak.proxy.upstream", None); + // SAFETY: the connection is live; it retains the queue and copies the block. + unsafe { + nw_connection_set_queue(conn.0, DispatchRetained::as_ptr(&queue).as_ptr().cast()); + nw_connection_set_state_changed_handler(conn.0, &on_state); + nw_connection_start(conn.0); + } + } + + match ready_rx.await { + Ok(Ok(())) => {} + Ok(Err(error)) => return Err(error.to_io()), + Err(_) => return Err(io::ErrorKind::ConnectionAborted.into()), + } + // With no direct route, "prefer" falls back to the system proxy, sending `CONNECT` to it. + // A proxy on this Mac may be this engine, which would tunnel it out the same way again, and + // again: the tunnel checks `is_own_fallback` first. Dropping `conn` cancels the connection + // before any request goes out. + let via_local_proxy = conn.used_local_proxy().await; + attempt.finish(via_local_proxy); + if via_local_proxy { + return Err(io::Error::other(format!( + "could not connect to {host}:{port} without the system proxy, which may be Yaak Proxy" + ))); + } + Ok(NwStream { + remote_addr: conn.remote_addr(), + conn, + shared: Arc::default(), + buffered: Bytes::new(), + eof: false, + read_error: None, + fin_sent: false, + }) +} + +/// Whether `host`, from a proxy endpoint, names this Mac. +fn is_local_host(host: &str) -> bool { + let host = host.trim_start_matches('[').trim_end_matches(']').trim_end_matches('.'); + host.eq_ignore_ascii_case("localhost") + || host.parse::().is_ok_and(|ip| ip.is_loopback()) +} + +/// Whether the proxy endpoint `endpoint` (an `nw_endpoint_t`) is on this Mac. One that can't be +/// read counts as local. Private Relay's proxies are remote hosts. +/// +/// # Safety +/// `endpoint` is a live `nw_endpoint_t`. +unsafe fn is_local_endpoint(endpoint: NwObject) -> bool { + // SAFETY: strings and addresses from the endpoint live as long as it does. + unsafe { + let text = |s: *const c_char| (!s.is_null()).then(|| CStr::from_ptr(s).to_string_lossy()); + match nw_endpoint_get_type(endpoint) { + ENDPOINT_TYPE_ADDRESS => socket_addr(nw_endpoint_get_address(endpoint)) + .is_none_or(|addr| addr.ip().is_loopback()), + ENDPOINT_TYPE_HOST => { + text(nw_endpoint_get_hostname(endpoint)).is_none_or(|h| is_local_host(&h)) + } + ENDPOINT_TYPE_URL => text(nw_endpoint_get_url(endpoint)) + .and_then(|url| url.parse::().ok()) + .and_then(|uri| uri.host().map(is_local_host)) + .unwrap_or(true), + _ => true, + } + } +} + +/// The address a plain-HTTP response came from, in its extensions. +#[derive(Clone, Copy, Debug)] +pub(crate) struct RemoteAddr(pub(crate) SocketAddr); + +/// Connects hyper-util's client to plain-HTTP origins through [`connect`]. +#[derive(Clone, Copy, Default)] +pub(crate) struct Connector; + +impl tower_service::Service for Connector { + type Response = NwIo; + type Error = io::Error; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, uri: Uri) -> Self::Future { + Box::pin(async move { + let invalid = |message| io::Error::new(io::ErrorKind::InvalidInput, message); + if uri.scheme() != Some(&Scheme::HTTP) { + return Err(invalid("invalid URL, scheme is not http")); + } + let host = uri.host().ok_or_else(|| invalid("invalid URL, host is missing"))?; + let host = host.trim_start_matches('[').trim_end_matches(']'); + let stream = connect(host, uri.port_u16().unwrap_or(80)).await?; + Ok(NwIo(TokioIo::new(stream))) + }) + } +} + +/// [`NwStream`] as hyper's IO, reporting where it connected. +pub(crate) struct NwIo(TokioIo); + +impl HyperConnection for NwIo { + fn connected(&self) -> Connected { + let connected = Connected::new(); + match self.0.inner().remote_addr() { + Some(addr) => connected.extra(RemoteAddr(addr)), + None => connected, + } + } +} + +impl hyper::rt::Read for NwIo { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: hyper::rt::ReadBufCursor<'_>, + ) -> Poll> { + Pin::new(&mut self.0).poll_read(cx, buf) + } +} + +impl hyper::rt::Write for NwIo { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.0).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.0).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.0).poll_shutdown(cx) + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::{TcpListener, TcpStream}; + + use super::*; + + const WAIT: Duration = Duration::from_secs(10); + + /// Byte `i` of a test payload: not periodic in any small power of two, so a dropped or + /// repeated chunk shows. + fn pattern(len: usize) -> Vec { + (0..len).map(|i| (i % 251) as u8).collect() + } + + async fn listen() -> (TcpListener, SocketAddr) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + (listener, addr) + } + + async fn accept(listener: &TcpListener) -> TcpStream { + tokio::time::timeout(WAIT, listener.accept()).await.expect("no connection").unwrap().0 + } + + /// Echoes one connection until the client closes its side, then closes. + fn spawn_echo(listener: TcpListener) { + tokio::spawn(async move { + let mut socket = accept(&listener).await; + let (mut read, mut write) = socket.split(); + tokio::io::copy(&mut read, &mut write).await.unwrap(); + write.shutdown().await.unwrap(); + }); + } + + #[tokio::test] + async fn echo_round_trip() { + let (listener, addr) = listen().await; + spawn_echo(listener); + + let mut stream = connect("127.0.0.1", addr.port()).await.unwrap(); + assert_eq!(stream.remote_addr(), Some(addr)); + stream.write_all(b"hello").await.unwrap(); + stream.flush().await.unwrap(); + let mut echo = [0u8; 5]; + stream.read_exact(&mut echo).await.unwrap(); + assert_eq!(&echo, b"hello"); + + stream.shutdown().await.unwrap(); + let mut rest = Vec::new(); + stream.read_to_end(&mut rest).await.unwrap(); + assert!(rest.is_empty()); + } + + #[tokio::test(flavor = "multi_thread")] + async fn large_transfer_round_trips() { + let (listener, addr) = listen().await; + spawn_echo(listener); + let payload = pattern(16 * 1024 * 1024); + + let stream = connect("127.0.0.1", addr.port()).await.unwrap(); + let (mut read, mut write) = tokio::io::split(stream); + let sent = payload.clone(); + let writer = tokio::spawn(async move { + for chunk in sent.chunks(100_000) { + write.write_all(chunk).await.unwrap(); + } + write.shutdown().await.unwrap(); + }); + // Read slowly at first, so the echo server's writes back up into ours. + let mut echoed = Vec::new(); + let mut buf = vec![0u8; 8192]; + for _ in 0..20 { + let n = read.read(&mut buf).await.unwrap(); + echoed.extend_from_slice(&buf[..n]); + tokio::time::sleep(Duration::from_millis(10)).await; + } + tokio::time::timeout(WAIT, read.read_to_end(&mut echoed)).await.unwrap().unwrap(); + writer.await.unwrap(); + assert_eq!(echoed.len(), payload.len()); + assert!(echoed == payload, "echoed bytes differ from the payload"); + } + + #[tokio::test(flavor = "multi_thread")] + async fn writes_wait_while_the_peer_is_not_reading() { + let (listener, addr) = listen().await; + let payload = pattern(64 * 1024 * 1024); + + let stream = connect("127.0.0.1", addr.port()).await.unwrap(); + let shared = stream.shared.clone(); + let mut server = accept(&listener).await; + let sent = payload.clone(); + let mut writer = tokio::spawn(async move { + let mut stream = stream; + for chunk in sent.chunks(100_000) { + stream.write_all(chunk).await.unwrap(); + } + stream.shutdown().await.unwrap(); + }); + + // The server reads nothing yet: once the socket buffers fill, the writer has to wait, + // with no more than a write's worth past the in-flight limit queued. + let stalled = tokio::time::timeout(Duration::from_millis(500), &mut writer).await; + assert!(stalled.is_err(), "64 MB went out with nobody reading"); + let queued = shared.write.lock().unwrap().bytes; + assert!(queued < MAX_IN_FLIGHT + 100_000, "{queued} bytes queued"); + + let mut received = Vec::new(); + tokio::time::timeout(WAIT, server.read_to_end(&mut received)).await.unwrap().unwrap(); + writer.await.unwrap(); + assert_eq!(received.len(), payload.len()); + assert!(received == payload, "received bytes differ from the payload"); + } + + #[tokio::test] + async fn server_half_close_leaves_the_write_side_open() { + let (listener, addr) = listen().await; + let (got_tx, got_rx) = oneshot::channel(); + tokio::spawn(async move { + let mut socket = accept(&listener).await; + socket.write_all(b"bye").await.unwrap(); + socket.shutdown().await.unwrap(); + let mut got = Vec::new(); + socket.read_to_end(&mut got).await.unwrap(); + let _ = got_tx.send(got); + }); + + let mut stream = connect("127.0.0.1", addr.port()).await.unwrap(); + let mut got = Vec::new(); + tokio::time::timeout(WAIT, stream.read_to_end(&mut got)).await.unwrap().unwrap(); + assert_eq!(got, b"bye"); + // Reads stay at end of stream. + assert_eq!(stream.read(&mut [0u8; 8]).await.unwrap(), 0); + + stream.write_all(b"still here").await.unwrap(); + stream.shutdown().await.unwrap(); + let server_got = tokio::time::timeout(WAIT, got_rx).await.unwrap().unwrap(); + assert_eq!(server_got, b"still here"); + let err = stream.write_all(b"late").await.unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::BrokenPipe); + } + + #[tokio::test] + async fn dropping_the_stream_closes_the_connection() { + let (listener, addr) = listen().await; + let stream = connect("127.0.0.1", addr.port()).await.unwrap(); + let mut server = accept(&listener).await; + + drop(stream); + let read = tokio::time::timeout(WAIT, server.read(&mut [0u8; 8])).await; + let read = read.expect("the connection stayed open after the stream was dropped"); + assert!(matches!(read, Ok(0) | Err(_)), "{read:?}"); + } + + #[tokio::test] + async fn dropping_a_pending_read_cancels_cleanly() { + let (listener, addr) = listen().await; + let mut stream = connect("127.0.0.1", addr.port()).await.unwrap(); + let mut server = accept(&listener).await; + + // A receive is still outstanding when the stream goes, and completes after it. + let mut buf = [0u8; 8]; + let pending = tokio::time::timeout(Duration::from_millis(50), stream.read(&mut buf)); + assert!(pending.await.is_err()); + assert!(stream.shared.read.lock().unwrap().pending); + drop(stream); + let read = tokio::time::timeout(WAIT, server.read(&mut [0u8; 8])).await.unwrap(); + assert!(matches!(read, Ok(0) | Err(_)), "{read:?}"); + } + + #[tokio::test] + async fn refused_connection_is_an_error() { + let (listener, addr) = listen().await; + drop(listener); + + let err = tokio::time::timeout(WAIT, connect("127.0.0.1", addr.port())) + .await + .expect("a refused connection kept waiting") + .err() + .expect("connected to a closed port"); + assert_eq!(err.kind(), io::ErrorKind::ConnectionRefused, "{err}"); + } + + #[tokio::test] + async fn reset_by_the_server_is_a_read_error() { + let (listener, addr) = listen().await; + let mut stream = connect("127.0.0.1", addr.port()).await.unwrap(); + let server = accept(&listener).await; + server.set_zero_linger().unwrap(); + drop(server); + + let read = tokio::time::timeout(WAIT, stream.read(&mut [0u8; 8])).await.unwrap(); + let err = read.expect_err("a reset connection read cleanly"); + assert_eq!(err.kind(), io::ErrorKind::ConnectionReset, "{err}"); + } + + #[link(name = "Network", kind = "framework")] + unsafe extern "C" { + fn nw_privacy_context_create(description: *const c_char) -> NwObject; + fn nw_privacy_context_add_proxy(privacy_context: NwObject, proxy_config: NwObject); + fn nw_parameters_set_privacy_context(parameters: NwObject, privacy_context: NwObject); + fn nw_proxy_config_create_http_connect(endpoint: NwObject, tls: NwObject) -> NwObject; + fn nw_proxy_config_set_failover_allowed(proxy_config: NwObject, allowed: bool); + } + + /// Gives `parameters` an HTTP CONNECT proxy at `host:port`, as the system proxy would be, in a + /// privacy context of their own so no other connection in the process sees it. + fn use_proxy(parameters: NwObject, host: &str, port: u16) { + let host = CString::new(host).unwrap(); + let port = CString::new(port.to_string()).unwrap(); + // SAFETY: each object created here is released once, after the parameters retain it. + unsafe { + let endpoint = nw_endpoint_create_host(host.as_ptr(), port.as_ptr()); + let config = nw_proxy_config_create_http_connect(endpoint, ptr::null_mut()); + nw_proxy_config_set_failover_allowed(config, true); + let context = nw_privacy_context_create(c"yaak-proxy test".as_ptr()); + nw_privacy_context_add_proxy(context, config); + nw_parameters_set_privacy_context(parameters, context); + nw_release(context); + nw_release(config); + nw_release(endpoint); + } + } + + /// With no direct route to `target`, "prefer" sends `CONNECT` to the proxy at + /// `proxy_host`, here standing in for the engine. The connection is refused before a request + /// goes through it, and the proxy, asking at once as the engine's passthrough tunnel would, + /// learns that the `CONNECT` is the engine's own. + async fn assert_local_fallback_is_refused(target: &str, proxy_host: &str) { + let (proxy, proxy_addr) = listen().await; + let (seen_tx, seen_rx) = oneshot::channel(); + let own_target = target.to_string(); + tokio::spawn(async move { + let mut socket = accept(&proxy).await; + let mut head = Vec::new(); + while !head.ends_with(b"\r\n\r\n") { + head.push(socket.read_u8().await.unwrap()); + } + socket.write_all(b"HTTP/1.1 200 Connection established\r\n\r\n").await.unwrap(); + let own = is_own_fallback(&own_target, 80).await; + let mut after = Vec::new(); + let _ = socket.read_to_end(&mut after).await; + let _ = seen_tx.send((head, own, after)); + }); + + let port = proxy_addr.port(); + let opened = open(target, 80, |p| use_proxy(p, proxy_host, port)); + let err = tokio::time::timeout(WAIT, opened) + .await + .expect("the connection kept waiting") + .err() + .expect("connected through the local proxy"); + assert!(err.to_string().contains("without the system proxy"), "{err}"); + + let (head, own, after) = tokio::time::timeout(WAIT, seen_rx).await.unwrap().unwrap(); + let head = String::from_utf8_lossy(&head); + assert!(head.starts_with(&format!("CONNECT {target}:80 ")), "{head}"); + assert!(own, "the CONNECT wasn't recognized as the engine's own"); + assert!(after.is_empty(), "{} bytes went through the proxy", after.len()); + // Tunnels to the target stay refused for a while. + assert!(is_own_fallback(&target.to_uppercase(), 80).await); + } + + #[tokio::test] + async fn fallback_to_a_local_proxy_is_refused() { + assert_local_fallback_is_refused("yaak-proxy-test.invalid", "127.0.0.1").await; + } + + #[tokio::test] + async fn fallback_to_a_proxy_at_localhost_with_a_dot_is_refused() { + assert_local_fallback_is_refused("yaak-proxy-test-dot.invalid", "localhost.").await; + } + + #[tokio::test] + async fn direct_connection_is_not_an_own_fallback() { + let (listener, addr) = listen().await; + spawn_echo(listener); + let stream = connect("localhost", addr.port()).await.unwrap(); + assert!(!is_own_fallback("localhost", addr.port()).await); + drop(stream); + assert!(!is_own_fallback("localhost", addr.port()).await); + } + + /// Opens of the same target that never finish, perhaps other clients' to a slow host, hold a + /// new tunnel up for one wait in all, not one each. + #[tokio::test] + async fn pending_opens_are_waited_for_together() { + const HOST: &str = "yaak-proxy-test-pending.invalid"; + let guards: Vec<_> = (0..3).map(|_| AttemptGuard::start(HOST, 80)).collect(); + + let started = std::time::Instant::now(); + assert!(!is_own_fallback(HOST, 80).await); + let waited = started.elapsed(); + assert!(waited >= OWN_FALLBACK_WAIT, "{waited:?}"); + assert!(waited < OWN_FALLBACK_WAIT * 3 / 2, "{waited:?}"); + drop(guards); + } + + /// The first open through a local proxy answers, whatever the others are doing. + #[tokio::test] + async fn an_own_fallback_among_pending_opens_answers_at_once() { + const HOST: &str = "yaak-proxy-test-own-among-pending.invalid"; + let slow = [AttemptGuard::start(HOST, 80), AttemptGuard::start(HOST, 80)]; + let own = AttemptGuard::start(HOST, 80); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(50)).await; + own.finish(true); + }); + + let started = std::time::Instant::now(); + assert!(is_own_fallback(HOST, 80).await); + assert!(started.elapsed() < OWN_FALLBACK_WAIT / 2, "{:?}", started.elapsed()); + drop(slow); + } + + #[test] + fn local_proxy_hosts() { + for host in [ + "localhost", + "LOCALHOST", + "localhost.", + "127.0.0.1", + "127.8.9.1", + "::1", + "[::1]", + ] { + assert!(is_local_host(host), "{host}"); + } + for host in [ + "cp4.cloudflare.com", + "10.0.0.1", + "192.168.4.44", + "::ffff:8.8.8.8", + "", + ] { + assert!(!is_local_host(host), "{host}"); + } + } + + #[tokio::test] + async fn host_with_a_nul_is_invalid() { + let err = connect("exa\0mple.com", 80).await.err().unwrap(); + assert_eq!(err.kind(), io::ErrorKind::InvalidInput); + } +} diff --git a/crates/yaak-proxy/src/request.rs b/crates/yaak-proxy/src/request.rs index bdc7d5ed0..5bf8db271 100644 --- a/crates/yaak-proxy/src/request.rs +++ b/crates/yaak-proxy/src/request.rs @@ -17,6 +17,7 @@ use hyper::http::request::Parts; use hyper::http::uri::{Authority, PathAndQuery, Scheme}; use hyper::upgrade::OnUpgrade; use hyper::{Method, Request, Response, StatusCode, Uri}; +#[cfg(not(target_os = "macos"))] use hyper_util::client::legacy::connect::HttpInfo; use hyper_util::rt::{TokioExecutor, TokioIo}; use log::{debug, info, warn}; @@ -296,35 +297,60 @@ pub(crate) async fn handle_request( return Ok(handle_connect(req, client, engine)); } + // A request the engine sent that was routed back to it, as iCloud Private Relay does with + // port 80 while the system proxy is on: forwarding it again would loop forever. It is the + // engine's own copy of a flow already recorded, which gets this answer, so it records none. + if carries_via(&req, &engine) { + debug!("Refused {} {}: it looped back to Yaak Proxy", req.method(), req.uri()); + return Ok(Response::builder() + .status(StatusCode::LOOP_DETECTED) + .header(CONTENT_TYPE, "text/plain; charset=utf-8") + .body(text_body("request looped back to Yaak Proxy")) + .unwrap()); + } + // Some clients send a bare path (`POST /api`) as they would to the server itself. let from_host = req.uri().authority().is_none(); - let rejected = if from_host { url_from_host(&mut req, &engine).err() } else { None }; + let rejected = if from_host { url_from_host(&mut req).err() } else { None }; let flow = Flow::begin(engine.clone(), client, &req, req.uri().to_string()); if let Some((status, error)) = rejected { return Ok(flow.reject(status, error)); } - if from_host { - req.headers_mut().append(VIA, engine.via.clone()); - } + req.headers_mut().append(VIA, engine.via.clone()); let (parts, body) = req.into_parts(); let outgoing = Request::from_parts(parts, flow.capture_request(body)); Ok(match engine.http_client.request(outgoing).await { Ok(resp) => { - let remote_addr = resp.extensions().get::().map(|i| i.remote_addr()); + let remote_addr = plain_remote_addr(&resp); flow.respond(resp, remote_addr) } Err(e) => flow.fail(error_chain(&e)), }) } +/// The address `Engine::http_client` connected to for `resp`. +fn plain_remote_addr(resp: &Response) -> Option { + #[cfg(target_os = "macos")] + return resp.extensions().get::().map(|a| a.0); + #[cfg(not(target_os = "macos"))] + resp.extensions().get::().map(|i| i.remote_addr()) +} + +/// Whether `req` carries this engine's `Via` mark, which it adds to every plain-HTTP request it +/// forwards. +fn carries_via(req: &Request, engine: &Engine) -> bool { + let via = engine.via.to_str().unwrap_or_default(); + req.headers() + .get_all(VIA) + .iter() + .filter_map(|v| v.to_str().ok()) + .any(|v| v.split(',').any(|hop| hop.trim() == via)) +} + /// Gives a bare-path request the `http://` URL its `Host` names. Fails when there is no usable -/// `Host`, or when the request carries this engine's `Via` mark: then it is one the engine sent -/// that was routed back to it, and forwarding it again would loop forever. -fn url_from_host( - req: &mut Request, - engine: &Engine, -) -> Result<(), (StatusCode, &'static str)> { +/// `Host`. +fn url_from_host(req: &mut Request) -> Result<(), (StatusCode, &'static str)> { let Some(host) = req.headers().get(HOST) else { return Err((StatusCode::BAD_REQUEST, "no host in request")); }; @@ -339,17 +365,6 @@ fn url_from_host( } *req.uri_mut() = Uri::from_parts(parts).map_err(|_| (StatusCode::BAD_REQUEST, "invalid host in request"))?; - - let via = engine.via.to_str().unwrap_or_default(); - let looped = req - .headers() - .get_all(VIA) - .iter() - .filter_map(|v| v.to_str().ok()) - .any(|v| v.split(',').any(|hop| hop.trim() == via)); - if looped { - return Err((StatusCode::LOOP_DETECTED, "request looped back to Yaak Proxy")); - } Ok(()) } diff --git a/crates/yaak-proxy/src/tunnel.rs b/crates/yaak-proxy/src/tunnel.rs index 5fe5a93d9..c27577ac4 100644 --- a/crates/yaak-proxy/src/tunnel.rs +++ b/crates/yaak-proxy/src/tunnel.rs @@ -8,13 +8,43 @@ use hyper_util::rt::TokioIo; use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf, copy_bidirectional}; use tokio::net::TcpStream; -/// Copies bytes both ways between `client` and `target`, after sending `prefix` (bytes already -/// read from the client) upstream. -pub(crate) async fn tunnel(mut client: S, target: &str, prefix: &[u8]) -> io::Result<()> +/// Copies bytes both ways between `client` and `target` (`host:port`), after sending `prefix` +/// (bytes already read from the client) upstream. +pub(crate) async fn tunnel(client: S, target: &str, prefix: &[u8]) -> io::Result<()> where S: AsyncRead + AsyncWrite + Unpin, { - let mut upstream = TcpStream::connect(target).await?; + // Private Relay sends public port-80 connections into the system proxy, which may be this + // engine. + #[cfg(target_os = "macos")] + if let Some((host, "80")) = target.rsplit_once(':') { + return tunnel_skipping_proxy(client, host, 80, prefix).await; + } + relay(client, TcpStream::connect(target).await?, prefix).await +} + +/// [`tunnel`] through Network.framework, which can skip the system proxy. When it can't, it falls +/// back to the system proxy with a `CONNECT` that arrives here, and tunneling that out the same +/// way would repeat it. +#[cfg(target_os = "macos")] +async fn tunnel_skipping_proxy(client: S, host: &str, port: u16, prefix: &[u8]) -> io::Result<()> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let host = host.trim_start_matches('[').trim_end_matches(']'); + if crate::nw::is_own_fallback(host, port).await { + return Err(io::Error::other(format!( + "not tunneling Yaak Proxy's own connection to {host}:{port} back out" + ))); + } + relay(client, crate::nw::connect(host, port).await?, prefix).await +} + +async fn relay(mut client: S, mut upstream: U, prefix: &[u8]) -> io::Result<()> +where + S: AsyncRead + AsyncWrite + Unpin, + U: AsyncRead + AsyncWrite + Unpin, +{ upstream.write_all(prefix).await?; copy_bidirectional(&mut client, &mut upstream).await?; Ok(()) @@ -83,3 +113,50 @@ impl AsyncWrite for Rewind { Pin::new(&mut self.inner).poll_shutdown(cx) } } + +#[cfg(all(test, target_os = "macos"))] +mod tests { + use std::time::Duration; + + use tokio::io::{AsyncReadExt, duplex}; + use tokio::net::TcpListener; + + use super::*; + + /// A port-80 tunnel on macOS goes through Network.framework: the prefix goes first, bytes + /// cross both ways, and each side's close reaches the other. + #[tokio::test(flavor = "multi_thread")] + async fn tunnel_skipping_proxy_relays_both_ways_and_closes() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let (mut read, mut write) = socket.split(); + tokio::io::copy(&mut read, &mut write).await.unwrap(); + write.shutdown().await.unwrap(); + }); + + let (client, proxy_side) = duplex(1024); + let tunnel = tokio::spawn(async move { + tunnel_skipping_proxy(proxy_side, "localhost", port, b"first,").await + }); + + let payload: Vec = (0..300_000).map(|i| (i % 251) as u8).collect(); + let (mut read, mut write) = tokio::io::split(client); + let (echoed, ()) = tokio::join!( + async { + let mut echoed = Vec::new(); + read.read_to_end(&mut echoed).await.unwrap(); + echoed + }, + async { + write.write_all(&payload).await.unwrap(); + write.shutdown().await.unwrap(); + }, + ); + assert_eq!(&echoed[..6], b"first,"); + assert!(echoed[6..] == payload[..], "echoed bytes differ from the payload"); + let ended = tokio::time::timeout(Duration::from_secs(10), tunnel).await; + ended.expect("the tunnel stayed open").unwrap().unwrap(); + } +} diff --git a/crates/yaak-proxy/tests/own_fallback.rs b/crates/yaak-proxy/tests/own_fallback.rs new file mode 100644 index 000000000..786132aa0 --- /dev/null +++ b/crates/yaak-proxy/tests/own_fallback.rs @@ -0,0 +1,118 @@ +//! The engine's own upstream falling back to the system proxy, which is the engine. Its own test +//! binary, since the proxy is set for the whole process. +#![cfg(target_os = "macos")] + +use std::ffi::{CString, c_void}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +use rustls::pki_types::ServerName; +use rustls::{ClientConfig, RootCertStore}; +use tempfile::TempDir; +use tokio::io::{AsyncReadExt, AsyncWriteExt, copy_bidirectional}; +use tokio::net::{TcpListener, TcpStream}; +use tokio_rustls::TlsConnector; +use yaak_proxy::{CaStore, ProxyConfig, start_proxy}; + +type NwObject = *mut c_void; + +#[link(name = "Network", kind = "framework")] +unsafe extern "C" { + static _nw_privacy_context_default_context: NwObject; + fn nw_endpoint_create_host(host: *const i8, port: *const i8) -> NwObject; + fn nw_proxy_config_create_http_connect(endpoint: NwObject, tls: NwObject) -> NwObject; + fn nw_proxy_config_set_failover_allowed(proxy_config: NwObject, allowed: bool); + fn nw_privacy_context_add_proxy(privacy_context: NwObject, proxy_config: NwObject); + fn nw_release(obj: NwObject); +} + +/// Points every Network.framework connection in the process at an HTTP CONNECT proxy on +/// `127.0.0.1:port`, as a system proxy would. +fn set_process_proxy(port: u16) { + let host = CString::new("127.0.0.1").unwrap(); + let port = CString::new(port.to_string()).unwrap(); + // SAFETY: each object created here is released once, after the context retains it. + unsafe { + let endpoint = nw_endpoint_create_host(host.as_ptr(), port.as_ptr()); + let config = nw_proxy_config_create_http_connect(endpoint, std::ptr::null_mut()); + nw_proxy_config_set_failover_allowed(config, true); + nw_privacy_context_add_proxy(_nw_privacy_context_default_context, config); + nw_release(config); + nw_release(endpoint); + } +} + +async fn read_head(stream: &mut (impl AsyncReadExt + Unpin)) -> String { + let mut head = Vec::new(); + while !head.ends_with(b"\r\n\r\n") { + head.push(stream.read_u8().await.unwrap()); + } + String::from_utf8(head).unwrap() +} + +/// A host the engine passes through is tunneled as soon as its `CONNECT` arrives. When that +/// `CONNECT` is the engine's own upstream falling back to the proxy, tunneling it out again would +/// fall back again, without end. One `CONNECT`, then the client gets a 502. +#[tokio::test(flavor = "multi_thread")] +async fn own_fallback_through_a_passthrough_host_stops() { + const TARGET: &str = "yaak-proxy-own-fallback.invalid"; + let dir = TempDir::new().unwrap(); + let ca = Arc::new(CaStore::load_or_create(dir.path()).unwrap()); + let proxy = start_proxy(ProxyConfig::new(0, ca)).unwrap(); + let engine_port = proxy.port; + + // Counts the CONNECTs on their way from Network.framework to the engine. + let gateway = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let gateway_port = gateway.local_addr().unwrap().port(); + let connects = Arc::new(AtomicUsize::new(0)); + let seen = connects.clone(); + tokio::spawn(async move { + loop { + let (mut inbound, _) = gateway.accept().await.unwrap(); + let seen = seen.clone(); + tokio::spawn(async move { + let head = read_head(&mut inbound).await; + if head.starts_with("CONNECT ") { + seen.fetch_add(1, Ordering::SeqCst); + } + let mut engine = TcpStream::connect(("127.0.0.1", engine_port)).await.unwrap(); + engine.write_all(head.as_bytes()).await.unwrap(); + let _ = copy_bidirectional(&mut inbound, &mut engine).await; + }); + } + }); + set_process_proxy(gateway_port); + + // A client that rejects the engine's certificate gets the target passed through. + let mut tcp = TcpStream::connect(("127.0.0.1", engine_port)).await.unwrap(); + let connect = format!("CONNECT {TARGET}:80 HTTP/1.1\r\nHost: {TARGET}:80\r\n\r\n"); + tcp.write_all(connect.as_bytes()).await.unwrap(); + assert!(read_head(&mut tcp).await.starts_with("HTTP/1.1 200")); + let distrusting = + ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + .with_safe_default_protocol_versions() + .unwrap() + .with_root_certificates(RootCertStore::empty()) + .with_no_client_auth(); + let name = ServerName::try_from(TARGET).unwrap(); + assert!(TlsConnector::from(Arc::new(distrusting)).connect(name, tcp).await.is_err()); + let addr = format!("{TARGET}:80"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + while proxy.handshake_state(&addr) != (0, true) { + assert!(tokio::time::Instant::now() < deadline, "{addr} was never passed through"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + + let mut client = TcpStream::connect(("127.0.0.1", engine_port)).await.unwrap(); + let request = format!("GET http://{TARGET}/ HTTP/1.1\r\nHost: {TARGET}\r\n\r\n"); + client.write_all(request.as_bytes()).await.unwrap(); + let head = tokio::time::timeout(Duration::from_secs(10), read_head(&mut client)) + .await + .expect("no response"); + assert!(head.starts_with("HTTP/1.1 502"), "{head}"); + + // Give a runaway chain time to show. + tokio::time::sleep(Duration::from_secs(1)).await; + assert_eq!(connects.load(Ordering::SeqCst), 1); +} diff --git a/crates/yaak-proxy/tests/proxy.rs b/crates/yaak-proxy/tests/proxy.rs index 322bf35dc..7e045d70e 100644 --- a/crates/yaak-proxy/tests/proxy.rs +++ b/crates/yaak-proxy/tests/proxy.rs @@ -1449,9 +1449,10 @@ async fn plain_http_bare_path_without_host_is_a_bad_request() { assert!(!events.iter().any(|e| matches!(e, ProxyEvent::ResponseStart { .. }))); } -/// A request whose connection upstream is routed back into the proxy, as happened to Telegram's -/// while the system proxy was on, arrives again with a bare path. Sending that on would loop -/// forever. +/// A request whose connection upstream is routed back into the proxy, as Private Relay does with +/// port 80 while the system proxy is on, arrives again with the engine's `Via` mark. Sending +/// that on would loop forever. It is refused without a flow of its own, so the loop shows as the +/// client's request answered with a 508. #[tokio::test(flavor = "multi_thread")] async fn plain_http_request_looped_back_to_the_proxy_stops() { let h = Harness::start(1024); @@ -1462,8 +1463,33 @@ async fn plain_http_request_looped_back_to_the_proxy_stops() { assert_eq!(resp.text().await.unwrap(), "request looped back to Yaak Proxy"); let events = h.flow_events(); - let error = find!(events, ProxyEvent::Error { error, .. } => error.clone()); - assert_eq!(error, "request looped back to Yaak Proxy"); + let starts: Vec<_> = events + .iter() + .filter_map(|e| match e { + ProxyEvent::RequestStart { url, .. } => Some(url.clone()), + _ => None, + }) + .collect(); + assert_eq!(starts, [url]); + assert_eq!(find!(events, ProxyEvent::ResponseStart { status, .. } => *status), 508); + assert!(!events.iter().any(|e| matches!(e, ProxyEvent::Error { .. }))); + let (body, _) = body(&events, false); + assert_eq!(body, b"request looped back to Yaak Proxy"); +} + +/// The same loop from a client that sends a bare path, as Telegram did. +#[tokio::test(flavor = "multi_thread")] +async fn bare_path_request_looped_back_to_the_proxy_stops() { + let h = Harness::start(1024); + let host = format!("127.0.0.1:{}", h.proxy.port); + + let resp = h.send_as_written(telegram_request(Some(&host), "/api", &[0xefu8; 44])).await; + assert_eq!(resp.status(), 508); + + let events = h.flow_events(); + let flows: HashSet = events.iter().map(flow_id).collect(); + assert_eq!(flows.len(), 1, "{events:?}"); + assert_eq!(find!(events, ProxyEvent::ResponseStart { status, .. } => *status), 508); } // -- HTTP/2 clients --