From ea288ec07415c43f056293e99c8191c11fd58a10 Mon Sep 17 00:00:00 2001 From: Gregory Schier Date: Sun, 15 Nov 2026 09:25:53 -0700 Subject: [PATCH] Yaak Proxy: dial the server before answering a CONNECT The engine now opens the TCP connection to a CONNECT's server before it answers. When the server can't be reached, the CONNECT gets a 502 (or a 504 on a timeout) and the client reports its own connection error, rather than a 200 followed by a 502 inside the decrypted session or a silent hang-up in a tunnel. The failure is recorded as one error flow. The dialled connection is handed to the tunnel, or to the first upstream connection of a decrypted one, so nothing is dialled twice. Co-Authored-By: Claude Opus 5.5 --- apps/yaak-proxy/harness/httpsFixtures.ts | 15 +- apps/yaak-proxy/lib/flowActions.ts | 2 +- crates-proxy/yaak-proxy-lib/src/lib.rs | 41 ++- crates-proxy/yaak-proxy-lib/src/models.rs | 3 +- crates-proxy/yaak-proxy-lib/tests/control.rs | 12 +- .../yaak-proxy-lib/tests/decryption.rs | 51 ++- crates-proxy/yaak-proxy-lib/tests/flow.rs | 13 +- crates/yaak-proxy/src/dial.rs | 298 ++++++++++++++++++ crates/yaak-proxy/src/lib.rs | 26 +- crates/yaak-proxy/src/nw.rs | 30 +- crates/yaak-proxy/src/request.rs | 222 ++++++++++--- crates/yaak-proxy/src/tunnel.rs | 3 +- crates/yaak-proxy/tests/own_fallback.rs | 41 +-- crates/yaak-proxy/tests/proxy.rs | 238 +++++++++++++- 14 files changed, 889 insertions(+), 106 deletions(-) create mode 100644 crates/yaak-proxy/src/dial.rs diff --git a/apps/yaak-proxy/harness/httpsFixtures.ts b/apps/yaak-proxy/harness/httpsFixtures.ts index 1a67d60d5..fc9d448a7 100644 --- a/apps/yaak-proxy/harness/httpsFixtures.ts +++ b/apps/yaak-proxy/harness/httpsFixtures.ts @@ -4,7 +4,7 @@ import { fixtures } from "./fixtures"; /** * Dev only: traffic for HTTPS decryption's states, by `?scene=`: `encrypted`, the first minute * with nothing decrypted, and `mixed`, some decrypted, a refused handshake and the tunnel passed - * through after it. + * through after it, and a server that could not be reached. */ export type HttpsScene = "encrypted" | "mixed"; @@ -169,6 +169,19 @@ function mixedScene(): Flow[] { }), decrypted(9, 0, "https://api.typefully.com/v1/drafts", apps.dia), tunnel(10, "www.google.com", apps.dia, { elapsedMs: null }), + { + ...tunnel(11, "jig3d.com", apps.dia), + method: "", + url: "https://jig3d.com", + tunnelled: false, + tunnelReason: null, + error: + "DNS lookup failed: failed to lookup address information: nodename nor servname provided, or not known", + elapsedMs: 3, + remoteAddr: null, + requestSize: 0, + responseSize: 0, + }, ]; } diff --git a/apps/yaak-proxy/lib/flowActions.ts b/apps/yaak-proxy/lib/flowActions.ts index 495ce1fc2..6727a8050 100644 --- a/apps/yaak-proxy/lib/flowActions.ts +++ b/apps/yaak-proxy/lib/flowActions.ts @@ -141,7 +141,7 @@ export function useFlowActions(): (flow: Flow) => DropdownItem[] { } }; const noResponse = (flow.error != null || flow.cancelled) && flow.status == null; - // A failed TLS handshake, recorded without a request, has nothing to replay. + // A failed TLS handshake or connection, recorded without a request, has nothing to replay. const noRequest = flow.method === ""; return [ { diff --git a/crates-proxy/yaak-proxy-lib/src/lib.rs b/crates-proxy/yaak-proxy-lib/src/lib.rs index 43a089284..7d37118e1 100644 --- a/crates-proxy/yaak-proxy-lib/src/lib.rs +++ b/crates-proxy/yaak-proxy-lib/src/lib.rs @@ -3226,8 +3226,8 @@ const STOPPED_MID_STREAM: &str = "Yaak Proxy stopped before the stream ended"; /// Why a tunnel still open when the engine stopped ended. const STOPPED_MID_TUNNEL: &str = "Yaak Proxy stopped before the connection closed"; -/// Why a failed handshake's flow can't be saved as a request. -pub(crate) const NOT_A_REQUEST: &str = "The TLS handshake failed before any request was sent"; +/// Why a failed handshake's or connection's flow can't be saved as a request. +pub(crate) const NOT_A_REQUEST: &str = "The connection failed before any request was sent"; /// The error of a flow recording a client's refusal of the proxy's certificate. const REJECTED_CERTIFICATE: &str = "Certificate rejected by the app"; @@ -3729,6 +3729,43 @@ impl FlowSink { graphql_scan: None, }); } + // No tunnel and no handshake: a failed connection, of the one event. + ProxyEvent::ConnectFailed { + id: _, + started_at, + authority, + connection, + client_addr, + error, + elapsed_ms, + } => { + if !recording.lock().map(|r| r.was_on_at(started_at)).unwrap_or(true) { + return; + } + attribution.begin(connection, client_addr); + let flow = Flow { + id: Flow::generate_id(), + started_at: DateTime::::from(started_at).naive_utc(), + url: format!("https://{authority}"), + client_http_version: "HTTP/1.1".to_string(), + client_addr: Some(client_addr.to_string()), + elapsed_ms: Some(clamp_ms(elapsed_ms)), + error: Some(error), + ..Default::default() + }; + self.finish_flow(InFlight { + flow, + connection, + request_body: None, + response_body: None, + persisted: false, + cleared: false, + next_seq: 0, + dirty: false, + last_write: None, + graphql_scan: None, + }); + } ProxyEvent::StreamStart { id, kind } => { if let Some(f) = in_flight.get_mut(&id) { let kind = StreamKind::from(kind); diff --git a/crates-proxy/yaak-proxy-lib/src/models.rs b/crates-proxy/yaak-proxy-lib/src/models.rs index 7aab26173..b3446a4bf 100644 --- a/crates-proxy/yaak-proxy-lib/src/models.rs +++ b/crates-proxy/yaak-proxy-lib/src/models.rs @@ -229,7 +229,8 @@ impl Flow { } } - /// A client's failed TLS handshake, recorded without a request: nothing to save or replay. + /// A client's failed TLS handshake, or a server that could not be reached, recorded without + /// a request: nothing to save or replay. pub fn is_handshake_only(&self) -> bool { self.method.is_empty() } diff --git a/crates-proxy/yaak-proxy-lib/tests/control.rs b/crates-proxy/yaak-proxy-lib/tests/control.rs index dabbfdabe..388f8e69b 100644 --- a/crates-proxy/yaak-proxy-lib/tests/control.rs +++ b/crates-proxy/yaak-proxy-lib/tests/control.rs @@ -933,15 +933,21 @@ async fn the_decrypt_list_reaches_the_cli_that_owns_the_engine() { let list = rpc(&app, "set_decrypt_list", json!({"entries": ["*"]})).await.unwrap(); assert_eq!(list["entries"][0], json!({"entry": "*", "builtIn": false})); - // Decrypted, the request is a flow of its own: a 502, as nothing listens there. + // Decrypted, the request is a flow of its own: a 502, as the server hangs up on its TLS. let ca = std::fs::read(dir.path().join("ca/ca.pem")).unwrap(); - let closed = TcpListener::bind("127.0.0.1:0").await.unwrap().local_addr().unwrap().port(); + let hangs_up = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = hangs_up.local_addr().unwrap().port(); + tokio::spawn(async move { + loop { + drop(hangs_up.accept().await.unwrap()); + } + }); let client = reqwest::Client::builder() .proxy(reqwest::Proxy::all(format!("http://127.0.0.1:{}", cli.port().unwrap())).unwrap()) .add_root_certificate(reqwest::Certificate::from_pem(&ca).unwrap()) .build() .unwrap(); - let url = format!("https://localhost:{closed}/forwarded"); + let url = format!("https://localhost:{port}/forwarded"); assert_eq!(client.get(&url).send().await.unwrap().status(), 502); let deadline = Instant::now() + Duration::from_secs(5); loop { diff --git a/crates-proxy/yaak-proxy-lib/tests/decryption.rs b/crates-proxy/yaak-proxy-lib/tests/decryption.rs index a69af97fe..91323bbb9 100644 --- a/crates-proxy/yaak-proxy-lib/tests/decryption.rs +++ b/crates-proxy/yaak-proxy-lib/tests/decryption.rs @@ -169,7 +169,7 @@ async fn a_refused_handshake_is_one_flow_per_address() { assert!(rejected.elapsed_ms.is_some(), "ended, not pending"); // No request was sent, so there is none to save. let preview = rpc(&ctx, "preview_save", json!({"flowId": rejected.id})).await; - assert!(preview.unwrap_err().starts_with("The TLS handshake failed")); + assert!(preview.unwrap_err().starts_with("The connection failed")); // Passed through from now on, naming the refusal. let _tunnel = open_tunnel(&ctx, &target).await; @@ -241,6 +241,55 @@ async fn a_refused_handshake_is_one_flow_per_address() { } } +/// A server the engine can't reach fails the CONNECT, and is one failed flow: no tunnel, no +/// refused handshake, nothing to reconnect or save. +#[tokio::test(flavor = "multi_thread")] +async fn an_unreachable_server_is_one_failed_flow() { + // Everything is tunnelled, so a tunnel would be recorded, and kept to reconnect. + let (_dir, ctx, _events) = started(&[]); + let closed = TcpListener::bind("127.0.0.1:0").await.unwrap().local_addr().unwrap(); + let mut tcp = TcpStream::connect(("127.0.0.1", ctx.port().unwrap())).await.unwrap(); + let connect = format!("CONNECT {closed} HTTP/1.1\r\nHost: {closed}\r\n\r\n"); + tcp.write_all(connect.as_bytes()).await.unwrap(); + let mut head = Vec::new(); + while !head.ends_with(b"\r\n\r\n") { + head.push(tcp.read_u8().await.unwrap()); + } + assert!(head.starts_with(b"HTTP/1.1 502"), "{}", String::from_utf8_lossy(&head)); + + let url = format!("https://{closed}"); + let failed = wait_for(&ctx, "failed", |f| f.url == url).await; + assert_eq!(failed.method, ""); + assert!(failed.error.as_deref().unwrap().starts_with("Connection refused: "), "{failed:?}"); + assert!(failed.elapsed_ms.is_some(), "ended, not pending"); + assert_eq!(failed.status, None); + assert!(!failed.tunnelled); + assert_eq!((failed.tunnel_reason, failed.tls_rejected.as_deref()), (None, None)); + assert_eq!(failed.client_addr, Some(tcp.local_addr().unwrap().to_string())); + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(flows(&ctx).len(), 1, "one flow"); + + let ids = |listed: Value| -> Vec { + listed["flows"].as_array().unwrap().iter().map(|f| f["id"].clone()).collect() + }; + let errors = rpc(&ctx, "list_flows", json!({"query": "@is:error"})).await.unwrap(); + assert_eq!(ids(errors), vec![json!(failed.id)]); + for query in ["@is:encrypted", "@is:rejected"] { + let listed = rpc(&ctx, "list_flows", json!({"query": query})).await.unwrap(); + assert!(ids(listed).is_empty(), "{query}"); + } + + // Listed now, a tunnel to it would be decrypted on a reconnect; there is none. + let saved = rpc(&ctx, "edit_decrypt_list", json!({"add": ["127.0.0.1"]})).await.unwrap(); + assert_eq!(saved["stillEncrypted"], 0); + assert!(reconnectable(&ctx, json!({})).await.is_empty()); + assert!(reconnectable(&ctx, json!({"flowId": failed.id})).await.is_empty()); + let passthrough = rpc(&ctx, "list_passthrough", json!({})).await.unwrap(); + assert_eq!(passthrough["addresses"], json!([])); + let preview = rpc(&ctx, "preview_save", json!({"flowId": failed.id})).await; + assert!(preview.unwrap_err().starts_with("The connection failed")); +} + /// Open tunnels are kept whether or not flows are recorded; saving the list closes none of /// them but counts those it would decrypt, and reconnecting closes them by flow, by entry or /// all at once. diff --git a/crates-proxy/yaak-proxy-lib/tests/flow.rs b/crates-proxy/yaak-proxy-lib/tests/flow.rs index 5719e35f2..6a7be5f5b 100644 --- a/crates-proxy/yaak-proxy-lib/tests/flow.rs +++ b/crates-proxy/yaak-proxy-lib/tests/flow.rs @@ -281,8 +281,15 @@ async fn http_version_setting_applies_to_new_connections() { ctx.start().unwrap(); let port = ctx.port().unwrap(); let ca = std::fs::read(dir.path().join("ca/ca.pem")).unwrap(); - // Nothing listens there, so each flow is a 502, but its client side is spoken all the same. - let closed = TcpListener::bind("127.0.0.1:0").await.unwrap().local_addr().unwrap().port(); + // The server hangs up on its TLS, so each flow is a 502, but its client side is spoken all + // the same. + let hangs_up = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let server = hangs_up.local_addr().unwrap().port(); + tokio::spawn(async move { + loop { + drop(hangs_up.accept().await.unwrap()); + } + }); let mut sent = 0; let mut client_version = async || { @@ -293,7 +300,7 @@ async fn http_version_setting_applies_to_new_connections() { .add_root_certificate(reqwest::Certificate::from_pem(&ca).unwrap()) .build() .unwrap(); - let url = format!("https://localhost:{closed}/{sent}"); + let url = format!("https://localhost:{server}/{sent}"); assert_eq!(client.get(&url).send().await.unwrap().status(), 502); wait_for_flow(&ctx, &url).await.client_http_version }; diff --git a/crates/yaak-proxy/src/dial.rs b/crates/yaak-proxy/src/dial.rs new file mode 100644 index 000000000..f59998da2 --- /dev/null +++ b/crates/yaak-proxy/src/dial.rs @@ -0,0 +1,298 @@ +//! The connection to the server of a CONNECT, opened before the CONNECT is answered, so a server +//! that can't be reached fails the CONNECT itself and the client reports its own connection +//! error, as it would without the proxy. + +use std::io; +use std::net::SocketAddr; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::time::{Duration, Instant}; + +use hyper::StatusCode; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +use tokio::net::TcpStream; + +/// The longest the dial before answering a CONNECT may take, name lookup included. The system's +/// own limit is over a minute. +pub(crate) const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + +/// A TCP connection to a server: a socket, or on macOS for port 80 a Network.framework +/// connection, which skips the system proxy as [`crate::tunnel::tunnel`] does there. +pub(crate) enum UpstreamIo { + Tcp(TcpStream), + #[cfg(target_os = "macos")] + Nw(crate::nw::NwStream), +} + +impl UpstreamIo { + pub(crate) fn remote_addr(&self) -> Option { + match self { + Self::Tcp(tcp) => tcp.peer_addr().ok(), + #[cfg(target_os = "macos")] + Self::Nw(nw) => nw.remote_addr(), + } + } +} + +/// A connection opened as its CONNECT came in, and how long opening it took. +pub(crate) struct Dialled { + pub(crate) io: UpstreamIo, + pub(crate) took: Duration, +} + +/// Opens the connection to `host:port` for a CONNECT, within [`CONNECT_TIMEOUT`]. `host` is as +/// the CONNECT named it, an IPv6 address in brackets. +pub(crate) async fn dial(host: &str, port: u16) -> Result { + let start = Instant::now(); + let io = match tokio::time::timeout(CONNECT_TIMEOUT, open(host, port)).await { + Ok(io) => io?, + Err(_) => { + let e = io::Error::new( + io::ErrorKind::TimedOut, + format!("no connection after {} seconds", CONNECT_TIMEOUT.as_secs()), + ); + return Err(Unreachable::of(&e, false)); + } + }; + Ok(Dialled { io, took: start.elapsed() }) +} + +async fn open(host: &str, port: u16) -> Result { + #[cfg(target_os = "macos")] + if port == 80 { + let host = host.trim_start_matches('[').trim_end_matches(']'); + return crate::nw::connect(host, port) + .await + .map(UpstreamIo::Nw) + .map_err(|e| Unreachable::of(&e, crate::nw::is_lookup_failure(&e))); + } + // Looked up apart from connecting, so a failure is known to be the lookup's. + let addrs: Vec = tokio::net::lookup_host(format!("{host}:{port}")) + .await + .map_err(|e| Unreachable::of(&e, true))? + .collect(); + if addrs.is_empty() { + let e = io::Error::new(io::ErrorKind::NotFound, "no address for the host"); + return Err(Unreachable::of(&e, true)); + } + TcpStream::connect(&addrs[..]) + .await + .map(UpstreamIo::Tcp) + .map_err(|e| Unreachable::of(&e, false)) +} + +/// Whether a CONNECT to `host:port` may be one of the engine's own connections coming back +/// through the system proxy. Its loop guard can only tell once the CONNECT is answered, as the +/// open that sent it finishes then, so such a CONNECT is answered before its server is dialled, +/// and the tunnel's guard runs after. Only port 80 on macOS can be one. Does not wait. +pub(crate) fn may_be_own_connection(host: &str, port: u16) -> bool { + #[cfg(target_os = "macos")] + if port == 80 { + let host = host.trim_start_matches('[').trim_end_matches(']'); + return crate::nw::has_attempts(host, port); + } + let _ = (host, port); + false +} + +/// Why a CONNECT's server could not be reached: what is said of it, and the system's error. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct Unreachable { + pub(crate) kind: UnreachableKind, + pub(crate) error: String, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum UnreachableKind { + Lookup, + Refused, + HostUnreachable, + NetworkUnreachable, + TimedOut, + Other, +} + +impl Unreachable { + /// Classifies `e`, from looking the host up when `lookup` is set, or else from connecting. + pub(crate) fn of(e: &io::Error, lookup: bool) -> Self { + use io::ErrorKind::{ConnectionRefused, HostUnreachable, NetworkUnreachable, TimedOut}; + let kind = if lookup { + UnreachableKind::Lookup + } else { + match e.kind() { + ConnectionRefused => UnreachableKind::Refused, + HostUnreachable => UnreachableKind::HostUnreachable, + NetworkUnreachable => UnreachableKind::NetworkUnreachable, + TimedOut => UnreachableKind::TimedOut, + _ => UnreachableKind::Other, + } + }; + Self { kind, error: e.to_string() } + } + + /// The reason in plain words. + pub(crate) fn reason(&self) -> &'static str { + match self.kind { + UnreachableKind::Lookup => "DNS lookup failed", + UnreachableKind::Refused => "Connection refused", + UnreachableKind::HostUnreachable => "Host unreachable", + UnreachableKind::NetworkUnreachable => "Network unreachable", + UnreachableKind::TimedOut => "Timed out", + UnreachableKind::Other => "Connection failed", + } + } + + /// The CONNECT's answer: a gateway timeout for a timeout, a bad gateway otherwise. + pub(crate) fn status(&self) -> StatusCode { + match self.kind { + UnreachableKind::TimedOut => StatusCode::GATEWAY_TIMEOUT, + _ => StatusCode::BAD_GATEWAY, + } + } +} + +/// The reason, then the system's error. +impl std::fmt::Display for Unreachable { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}: {}", self.reason(), self.error) + } +} + +impl AsyncRead for UpstreamIo { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + match self.get_mut() { + Self::Tcp(s) => Pin::new(s).poll_read(cx, buf), + #[cfg(target_os = "macos")] + Self::Nw(s) => Pin::new(s).poll_read(cx, buf), + } + } +} + +impl AsyncWrite for UpstreamIo { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + match self.get_mut() { + Self::Tcp(s) => Pin::new(s).poll_write(cx, buf), + #[cfg(target_os = "macos")] + Self::Nw(s) => Pin::new(s).poll_write(cx, buf), + } + } + + /// Passed through, so TLS records leave in one write rather than one each, which Nagle's + /// algorithm would hold for the server's delayed ACK. + fn poll_write_vectored( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + bufs: &[io::IoSlice<'_>], + ) -> Poll> { + match self.get_mut() { + Self::Tcp(s) => Pin::new(s).poll_write_vectored(cx, bufs), + #[cfg(target_os = "macos")] + Self::Nw(s) => Pin::new(s).poll_write_vectored(cx, bufs), + } + } + + fn is_write_vectored(&self) -> bool { + match self { + Self::Tcp(s) => s.is_write_vectored(), + #[cfg(target_os = "macos")] + Self::Nw(s) => s.is_write_vectored(), + } + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match self.get_mut() { + Self::Tcp(s) => Pin::new(s).poll_flush(cx), + #[cfg(target_os = "macos")] + Self::Nw(s) => Pin::new(s).poll_flush(cx), + } + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match self.get_mut() { + Self::Tcp(s) => Pin::new(s).poll_shutdown(cx), + #[cfg(target_os = "macos")] + Self::Nw(s) => Pin::new(s).poll_shutdown(cx), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn classify(kind: io::ErrorKind, lookup: bool) -> (UnreachableKind, &'static str, u16) { + let unreachable = Unreachable::of(&io::Error::new(kind, "system error"), lookup); + (unreachable.kind, unreachable.reason(), unreachable.status().as_u16()) + } + + #[test] + fn failures_are_classified_by_kind() { + use UnreachableKind as K; + use io::ErrorKind::*; + assert_eq!(classify(ConnectionRefused, false), (K::Refused, "Connection refused", 502)); + assert_eq!(classify(HostUnreachable, false), (K::HostUnreachable, "Host unreachable", 502)); + assert_eq!( + classify(NetworkUnreachable, false), + (K::NetworkUnreachable, "Network unreachable", 502) + ); + assert_eq!(classify(TimedOut, false), (K::TimedOut, "Timed out", 504)); + assert_eq!(classify(ConnectionReset, false), (K::Other, "Connection failed", 502)); + } + + /// Whatever the lookup's error looks like, it failed to find the host. + #[test] + fn a_lookup_failure_is_dns_whatever_its_kind() { + for kind in [ + io::ErrorKind::Other, + io::ErrorKind::NotFound, + io::ErrorKind::TimedOut, + ] { + assert_eq!(classify(kind, true), (UnreachableKind::Lookup, "DNS lookup failed", 502)); + } + } + + #[test] + fn the_text_is_the_reason_then_the_systems_error() { + let e = io::Error::from(io::ErrorKind::ConnectionRefused); + let text = Unreachable::of(&e, false).to_string(); + assert_eq!(text, format!("Connection refused: {e}")); + } + + /// A name under `.invalid` never resolves (RFC 2606). + #[tokio::test] + async fn an_invalid_name_is_a_dns_failure() { + let Err(unreachable) = dial("yaak-proxy-dial.invalid", 443).await else { + panic!("an .invalid name resolved"); + }; + assert_eq!(unreachable.kind, UnreachableKind::Lookup, "{unreachable}"); + } + + /// Without vectored writes, TLS sends each record on its own, and a small one waits on + /// Nagle's algorithm for the server's delayed ACK: some 40 ms on every request. + #[tokio::test] + async fn writes_stay_vectored() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let dialled = dial("127.0.0.1", addr.port()).await.unwrap(); + assert!(dialled.io.is_write_vectored()); + } + + #[tokio::test] + async fn a_closed_port_is_refused() { + let closed = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = closed.local_addr().unwrap().port(); + drop(closed); + let Err(unreachable) = dial("127.0.0.1", port).await else { + panic!("connected to a closed port"); + }; + assert_eq!(unreachable.kind, UnreachableKind::Refused, "{unreachable}"); + } +} diff --git a/crates/yaak-proxy/src/lib.rs b/crates/yaak-proxy/src/lib.rs index 929fe25ff..844bd5de4 100644 --- a/crates/yaak-proxy/src/lib.rs +++ b/crates/yaak-proxy/src/lib.rs @@ -3,6 +3,7 @@ mod cert; mod connection; mod decode; mod decrypt; +mod dial; mod grpc; mod messages; #[cfg(target_os = "macos")] @@ -106,7 +107,8 @@ impl ProxyConfig { /// passes. An HTTPS connection the engine tunnels without decrypting, because the decrypt policy /// leaves it out or its client rejected the engine's certificate, is a flow of its own, from /// `TunnelStart` to `TunnelEnd`. A client's failed handshake that starts its address being -/// passed through is a flow too, of the one event `HandshakeFailed`. Other traffic tunnelled +/// passed through is a flow too, of the one event `HandshakeFailed`, and so is a CONNECT whose +/// server could not be reached, of the one event `ConnectFailed`. Other traffic tunnelled /// blind (non-TLS CONNECT payloads, upgrades to protocols other than WebSocket) produces no /// events. #[derive(Debug, Clone)] @@ -151,7 +153,11 @@ pub enum ProxyEvent { UpstreamConnection { id: u64, /// Milliseconds from `started_at` to the TCP connection, and to the TLS handshake's - /// end. None when the connection was opened for an earlier request and reused. + /// end. None when the connection was opened for an earlier request and reused. A + /// connection opened as its CONNECT came in, before this request, is charged to the + /// first request on it, as if opened just before it: `elapsed_connect_ms` is how long + /// opening it took, and this flow's later times are counted from that much before its + /// `started_at`, so they stay in order. elapsed_connect_ms: Option, elapsed_tls_ms: Option, /// `TLSv1.3`, `TLSv1.2`. @@ -265,6 +271,22 @@ pub enum ProxyEvent { /// Milliseconds from `started_at` to the failure. elapsed_ms: u64, }, + /// A CONNECT's server could not be reached, so the CONNECT was answered with a 502, or a 504 + /// when it timed out, and nothing was tunnelled or decrypted. A flow of its own, which + /// nothing follows. + ConnectFailed { + id: u64, + started_at: SystemTime, + /// `host[:port]`, the port left out when it is 443. + authority: String, + connection: u64, + client_addr: SocketAddr, + /// The reason in plain words, then the system's error: `DNS lookup failed: ...`, + /// `Connection refused: ...`, `Host unreachable: ...`, `Timed out: ...`. + error: String, + /// Milliseconds from `started_at` to the failure. + elapsed_ms: u64, + }, } /// Why a CONNECT is tunnelled rather than decrypted. diff --git a/crates/yaak-proxy/src/nw.rs b/crates/yaak-proxy/src/nw.rs index a5f901cbf..f8e7e003c 100644 --- a/crates/yaak-proxy/src/nw.rs +++ b/crates/yaak-proxy/src/nw.rs @@ -124,9 +124,7 @@ impl NwError { 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)) - } + ERROR_DOMAIN_DNS => io::Error::other(LookupFailed(self.code)), _ => io::Error::other(format!( "Network.framework error (domain {}, code {})", self.domain, self.code @@ -135,6 +133,23 @@ impl NwError { } } +/// A failed DNS lookup, by its `DNSServiceErrorType`. +#[derive(Debug)] +struct LookupFailed(i32); + +impl std::fmt::Display for LookupFailed { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "failed to look up address (DNS error {})", self.0) + } +} + +impl std::error::Error for LookupFailed {} + +/// Whether `e`, from [`connect`], is the host's name failing to resolve. +pub(crate) fn is_lookup_failure(e: &io::Error) -> bool { + e.get_ref().is_some_and(|inner| inner.is::()) +} + /// An `nw_connection_t`, cancelled when dropped. struct Connection(NwObject); @@ -538,6 +553,15 @@ pub(crate) async fn is_own_fallback(host: &str, port: u16) -> bool { false } +/// Whether an open of `host:port` is under way, or recently went through a proxy on this Mac: +/// whether a `CONNECT` to it may be the engine's own, which only [`is_own_fallback`] can tell, +/// once the `CONNECT` is answered. Does not wait. +pub(crate) fn has_attempts(host: &str, port: u16) -> bool { + let mut attempts = ATTEMPTS.lock().unwrap(); + prune_attempts(&mut attempts); + attempts.contains_key(&attempt_key(host, port)) +} + /// 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 diff --git a/crates/yaak-proxy/src/request.rs b/crates/yaak-proxy/src/request.rs index 3d6abe3c3..bb3cd0345 100644 --- a/crates/yaak-proxy/src/request.rs +++ b/crates/yaak-proxy/src/request.rs @@ -29,10 +29,11 @@ use tokio_rustls::{LazyConfigAcceptor, TlsConnector}; use crate::body::{BodyParser, CaptureBody, Direction}; use crate::connection::serve_tls; use crate::decrypt::Verdict; +use crate::dial::{self, Dialled, UpstreamIo}; use crate::grpc::{self, GrpcParser}; use crate::messages::{MessageDirection, StreamKind}; use crate::sse::SseParser; -use crate::tunnel::{Counted, Rewind, is_own_connection, splice, tunnel}; +use crate::tunnel::{Counted, Rewind, is_own_connection, relay, splice, tunnel}; use crate::ws::{self, WsFlow}; use crate::{ClientConn, Engine, ProxyEvent, TunnelReason}; @@ -191,15 +192,26 @@ impl Flow { } /// Records the connection the request goes on: its timings when it was opened for this - /// request, and its TLS either way. - fn goes_on(&self, conn: &UpstreamConn, reused: bool) { - let since = |at: Instant| { - (!reused).then(|| at.saturating_duration_since(self.start).as_millis() as u64) + /// request, and its TLS either way. One opened as the CONNECT came in is charged to its + /// first request as if opened just before it: the flow's times count from that much earlier. + fn goes_on(&mut self, conn: &UpstreamConn, reused: bool) { + let ms = |d: Duration| d.as_millis() as u64; + let (connect, tls) = if reused { + (None, None) + } else { + if let Some(took) = conn.tls.dialled_before { + self.start = self.start.checked_sub(took).unwrap_or(self.start); + } + let connect = match conn.tls.dialled_before { + Some(took) => took, + None => conn.tls.connected.saturating_duration_since(self.start), + }; + (Some(ms(connect)), Some(ms(conn.tls.handshaken.saturating_duration_since(self.start)))) }; self.engine.emit(ProxyEvent::UpstreamConnection { id: self.id, - elapsed_connect_ms: since(conn.tls.connected), - elapsed_tls_ms: since(conn.tls.handshaken), + elapsed_connect_ms: connect, + elapsed_tls_ms: tls, tls_version: conn.tls.version.clone(), tls_cipher: conn.tls.cipher.clone(), }); @@ -310,7 +322,7 @@ pub(crate) async fn handle_request( engine: Arc, ) -> Result, Infallible> { if req.method() == Method::CONNECT { - return Ok(handle_connect(req, client, engine)); + return Ok(handle_connect(req, client, engine).await); } // A request the engine sent that was routed back to it, as iCloud Private Relay does with @@ -395,7 +407,10 @@ struct Target { port: u16, } -fn handle_connect( +/// Answers a CONNECT once its server is reached, with the connection handed on to the tunnel or +/// the first request, or with a 502 or 504 when it can't be, so the client reports a connection +/// error of its own. +async fn handle_connect( req: Request, client: ClientConn, engine: Arc, @@ -408,19 +423,33 @@ fn handle_connect( }; let host = authority.host().to_string(); let port = authority.port_u16().unwrap_or(443); + let target = Target { + server_name: host.trim_start_matches('[').trim_end_matches(']').to_string(), + authority: if port == 443 { host.clone() } else { format!("{host}:{port}") }, + addr: format!("{host}:{port}"), + port, + }; + + // One that may be the engine's own connection coming back is answered first, as its loop + // guard can only tell once it is, and is dialled behind that guard. + let dialled = if dial::may_be_own_connection(&host, port) { + None + } else { + let (started_at, start) = (SystemTime::now(), Instant::now()); + match dial::dial(&host, port).await { + Ok(dialled) => Some(dialled), + Err(unreachable) => { + return refuse_connect(&engine, &target, client, &unreachable, started_at, start); + } + } + }; tokio::spawn(async move { let upgraded = match hyper::upgrade::on(req).await { Ok(u) => TokioIo::new(u), Err(e) => return warn!("CONNECT upgrade failed for {host}: {e}"), }; - let target = Target { - server_name: host.trim_start_matches('[').trim_end_matches(']').to_string(), - authority: if port == 443 { host.clone() } else { format!("{host}:{port}") }, - addr: format!("{host}:{port}"), - port, - }; - if let Err(e) = intercept(upgraded, target, client, engine).await { + if let Err(e) = intercept(upgraded, target, dialled, client, engine).await { debug!("Tunnel to {host}:{port} ended: {}", error_chain(e.as_ref())); } }); @@ -428,39 +457,72 @@ fn handle_connect( Response::new(text_body(Bytes::new())) } +/// Records a CONNECT whose server could not be reached as a flow of its own, and answers it. +fn refuse_connect( + engine: &Engine, + target: &Target, + conn: ClientConn, + unreachable: &dial::Unreachable, + started_at: SystemTime, + start: Instant, +) -> Response { + let error = unreachable.to_string(); + debug!("Could not connect to {}: {error}", target.addr); + engine.emit(ProxyEvent::ConnectFailed { + id: engine.next_flow_id(), + started_at, + authority: target.authority.clone(), + connection: conn.connection, + client_addr: conn.addr, + error: error.clone(), + elapsed_ms: start.elapsed().as_millis() as u64, + }); + Response::builder() + .status(unreachable.status()) + .header(CONTENT_TYPE, "text/plain; charset=utf-8") + .body(text_body(format!("Yaak Proxy could not connect to {}: {error}", target.addr))) + .unwrap() +} + +/// Judges an answered CONNECT and serves it: tunnelled, or decrypted. `dialled` is its server's +/// connection, opened before the CONNECT was answered; without it, the server is dialled here. async fn intercept( mut client: S, target: Target, + dialled: Option, conn: ClientConn, engine: Arc, ) -> Result<(), BoxError> where S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { - match engine.verdict(&target.server_name, target.port, conn.connection).await { - Verdict::Decrypt => {} + let reason = match engine.verdict(&target.server_name, target.port, conn.connection).await { + Verdict::Decrypt => None, Verdict::NotListed => { debug!("Tunneling {} undecrypted, as the decrypt list leaves it out", target.addr); - return Ok(tunnel_flow(client, &target, conn, &engine, TunnelReason::NotListed).await?); + Some(TunnelReason::NotListed) } Verdict::Excluded(entry) => { debug!("Tunneling {} undecrypted, as {entry} excludes it", target.addr); - let reason = - TunnelReason::Excluded { entry: entry.to_string(), built_in: entry.built_in }; - return Ok(tunnel_flow(client, &target, conn, &engine, reason).await?); + Some(TunnelReason::Excluded { entry: entry.to_string(), built_in: entry.built_in }) } - } + }; // The fallback for clients that rejected the engine's certificate. - if let Some(rejected_by) = engine.passthrough_of(&target.addr) { - let reason = TunnelReason::Rejected { rejected_by }; - return Ok(tunnel_flow(client, &target, conn, &engine, reason).await?); + let reason = reason.or_else(|| { + engine + .passthrough_of(&target.addr) + .map(|rejected_by| TunnelReason::Rejected { rejected_by }) + }); + if let Some(reason) = reason { + let upstream = dialled.map(|d| d.io); + return Ok(tunnel_flow(client, &target, upstream, conn, &engine, reason).await?); } let mut buf = vec![0u8; 4096]; let Ok(n) = tokio::time::timeout(CLIENT_FIRST_BYTES_TIMEOUT, client.read(&mut buf)).await else { debug!("Client sent nothing, tunneling server-first traffic to {}", target.addr); - return Ok(tunnel(client, &target.addr, &[]).await?); + return Ok(tunnel_to(client, &target, dialled.map(|d| d.io), &[]).await?); }; let n = n?; if n == 0 { @@ -469,7 +531,7 @@ where buf.truncate(n); if buf[0] != TLS_HANDSHAKE_RECORD { debug!("Tunneling non-TLS traffic to {}", target.addr); - return Ok(tunnel(client, &target.addr, &buf).await?); + return Ok(tunnel_to(client, &target, dialled.map(|d| d.io), &buf).await?); } let attempt = HandshakeAttempt { started_at: SystemTime::now(), start: Instant::now() }; @@ -514,16 +576,35 @@ where let http2 = tls.get_ref().1.alpn_protocol() == Some(b"h2"); let tls = Rewind::new(Bytes::from(buf), tls); - let upstreams = Arc::new(Upstreams::new(Target { server_name, ..target })); + let upstreams = Arc::new(Upstreams::new(Target { server_name, ..target }, dialled)); serve_tls(tls, http2, move |req| forward_https(req, upstreams.clone(), conn, engine.clone())) .await } +/// Relays `client` to the server over `upstream`, the connection dialled before the CONNECT was +/// answered, or else over a new one, after sending `prefix` (bytes already read from the client). +async fn tunnel_to( + client: S, + target: &Target, + upstream: Option, + prefix: &[u8], +) -> std::io::Result<()> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + match upstream { + Some(upstream) => relay(client, upstream, prefix).await, + None => tunnel(client, &target.addr, prefix).await, + } +} + /// Tunnels a CONNECT undecrypted, recorded as a flow of its own with the bytes each way. Not -/// one of the engine's own connections coming back, which [`tunnel`] refuses. +/// one of the engine's own connections coming back, which [`tunnel`] refuses: one that may be +/// was answered without dialling, so has no `upstream`. async fn tunnel_flow( client: S, target: &Target, + upstream: Option, conn: ClientConn, engine: &Engine, reason: TunnelReason, @@ -531,7 +612,7 @@ async fn tunnel_flow( where S: AsyncRead + AsyncWrite + Unpin, { - if is_own_connection(&target.addr).await { + if upstream.is_none() && is_own_connection(&target.addr).await { return tunnel(client, &target.addr, &[]).await; } let id = engine.next_flow_id(); @@ -553,7 +634,7 @@ where let (sent, received) = client.counters(); // Closed on request: dropping the relay closes both sides. let (result, reconnect_requested) = tokio::select! { - result = tunnel(client, &target.addr, &[]) => (result, false), + result = tunnel_to(client, target, upstream, &[]) => (result, false), _ = close_rx => (Ok(()), true), }; engine.tunnels.lock().unwrap().remove(&id); @@ -613,6 +694,11 @@ fn record_handshake_failure( } } +/// The server closed the connection before or during the TLS handshake. +fn closed_by_server(e: &std::io::Error) -> bool { + is_hang_up(e) || e.kind() == std::io::ErrorKind::BrokenPipe +} + /// The client closed the connection without a TLS `close_notify`, or reset it. fn is_hang_up(e: &std::io::Error) -> bool { use std::io::ErrorKind::{ConnectionAborted, ConnectionReset, UnexpectedEof}; @@ -657,21 +743,29 @@ struct UpstreamConn { /// When a connection to the server was made, and the TLS it negotiated. #[derive(Debug, Clone)] struct TlsInfo { + /// When the TCP connection was made, or, for one opened as its CONNECT came in, when it was + /// taken up for TLS. connected: Instant, + /// How long opening it took, when it was opened as its CONNECT came in, before any request. + dialled_before: Option, handshaken: Instant, version: Option, cipher: Option, } impl TlsInfo { - fn of(conn: &rustls::ClientConnection, connected: Instant) -> Self { + fn of( + conn: &rustls::ClientConnection, + connected: Instant, + dialled_before: Option, + ) -> Self { let version = conn.protocol_version().map(|v| match v { rustls::ProtocolVersion::TLSv1_3 => "TLSv1.3".to_string(), rustls::ProtocolVersion::TLSv1_2 => "TLSv1.2".to_string(), other => format!("{other:?}"), }); let cipher = conn.negotiated_cipher_suite().map(|s| format!("{:?}", s.suite())); - Self { connected, handshaken: Instant::now(), version, cipher } + Self { connected, dialled_before, handshaken: Instant::now(), version, cipher } } } @@ -720,6 +814,8 @@ const MAX_IDLE_H1: usize = 6; /// them all at once; an HTTP/1.1 one a request at a time. struct Upstreams { target: Target, + /// The connection dialled before the CONNECT was answered, for the first one to the server. + dialled: std::sync::Mutex>, conns: tokio::sync::Mutex>, /// The server answered in HTTP/1.1, so each request in flight needs a connection of its /// own, and new ones need not wait to see whether the one before can be shared. @@ -727,8 +823,20 @@ struct Upstreams { } impl Upstreams { - fn new(target: Target) -> Self { - Self { target, conns: tokio::sync::Mutex::new(Vec::new()), h1: AtomicBool::new(false) } + fn new(target: Target, dialled: Option) -> Self { + Self { + target, + dialled: std::sync::Mutex::new(dialled), + conns: tokio::sync::Mutex::new(Vec::new()), + h1: AtomicBool::new(false), + } + } + + /// A new connection to the server: over the one dialled before the CONNECT was answered, + /// the first time, and dialled anew after. + async fn connect(&self, engine: &Engine, h1_only: bool) -> Result { + let dialled = self.dialled.lock().unwrap().take(); + connect_upstream(engine, &self.target, h1_only, dialled).await } /// A connection for one request, and whether it carried one before. Requests that arrive @@ -745,9 +853,9 @@ impl Upstreams { } if self.h1.load(Ordering::Relaxed) { drop(conns); - return Ok((connect_upstream(engine, &self.target, false).await?, false)); + return Ok((self.connect(engine, false).await?, false)); } - let conn = connect_upstream(engine, &self.target, false).await?; + let conn = self.connect(engine, false).await?; match &conn.sender { Upstream::H2(s) => conns.push(conn.share(Upstream::H2(s.clone()))), Upstream::H1(_) => self.h1.store(true, Ordering::Relaxed), @@ -772,18 +880,40 @@ impl Upstreams { } } +/// Opens a connection to the server for requests: over `dialled` when there is one, and over a +/// new TCP connection otherwise, or when `dialled` was closed while it waited. async fn connect_upstream( engine: &Engine, target: &Target, h1_only: bool, + dialled: Option, ) -> Result { - let tcp = TcpStream::connect(&target.addr).await?; - let connected = Instant::now(); - let remote_addr = tcp.peer_addr().ok(); let config = if h1_only { &engine.upstream_tls_h1 } else { &engine.upstream_tls }; + let connector = TlsConnector::from(config.clone()); let server_name = ServerName::try_from(target.server_name.clone())?; - let tls = TlsConnector::from(config.clone()).connect(server_name, tcp).await?; - let tls_info = TlsInfo::of(tls.get_ref().1, connected); + let mut tls = None; + if let Some(Dialled { io, took }) = dialled { + let (connected, remote_addr) = (Instant::now(), io.remote_addr()); + match connector.connect(server_name.clone(), io).await { + Ok(stream) => tls = Some((stream, connected, Some(took), remote_addr)), + // The server closed it while it waited for the first request. + Err(e) if closed_by_server(&e) => { + debug!("Connection to {} closed before its first request: {e}", target.addr); + } + Err(e) => return Err(e.into()), + } + } + let (tls, connected, dialled_before, remote_addr) = match tls { + Some(tls) => tls, + None => { + let tcp = TcpStream::connect(&target.addr).await?; + let connected = Instant::now(); + let remote_addr = tcp.peer_addr().ok(); + let stream = connector.connect(server_name, UpstreamIo::Tcp(tcp)).await?; + (stream, connected, None, remote_addr) + } + }; + let tls_info = TlsInfo::of(tls.get_ref().1, connected, dialled_before); let is_h2 = tls.get_ref().1.alpn_protocol() == Some(b"h2"); let io = TokioIo::new(tls); @@ -875,13 +1005,13 @@ async fn forward_https( let target = &upstreams.target; let path = req.uri().path_and_query().map(|pq| pq.to_string()).unwrap_or_else(|| "/".into()); let url = format!("https://{}{path}", target.authority); - let flow = Flow::begin(engine.clone(), client, &req, url); + let mut flow = Flow::begin(engine.clone(), client, &req, url); // Only HTTP/1.1 has `Upgrade`. Its connection is the tunnel afterwards, so it is not shared. let client_upgrade: Option = req.headers().contains_key(UPGRADE).then(|| hyper::upgrade::on(&mut req)); let taken = match client_upgrade { - Some(_) => connect_upstream(&engine, target, true).await.map(|c| (c, false)), + Some(_) => upstreams.connect(&engine, true).await.map(|c| (c, false)), None => upstreams.take(&engine).await, }; let (mut conn, mut reused) = match taken { @@ -907,7 +1037,7 @@ async fn forward_https( return Ok(flow.fail(error_chain(e.error()))); }; debug!("Upstream connection to {} was closed, retrying on a new one", target.addr); - conn = match connect_upstream(&engine, target, false).await { + conn = match upstreams.connect(&engine, false).await { Ok(conn) => conn, Err(e) => return Ok(flow.fail(error_chain(e.as_ref()))), }; diff --git a/crates/yaak-proxy/src/tunnel.rs b/crates/yaak-proxy/src/tunnel.rs index ad77a2ff0..5d914ef88 100644 --- a/crates/yaak-proxy/src/tunnel.rs +++ b/crates/yaak-proxy/src/tunnel.rs @@ -108,7 +108,8 @@ where relay(client, crate::nw::connect(host, port).await?, prefix).await } -async fn relay(mut client: S, mut upstream: U, prefix: &[u8]) -> io::Result<()> +/// Copies bytes both ways between `client` and `upstream`, after sending `prefix` upstream. +pub(crate) async fn relay(mut client: S, mut upstream: U, prefix: &[u8]) -> io::Result<()> where S: AsyncRead + AsyncWrite + Unpin, U: AsyncRead + AsyncWrite + Unpin, diff --git a/crates/yaak-proxy/tests/own_fallback.rs b/crates/yaak-proxy/tests/own_fallback.rs index 786132aa0..b481a0ba2 100644 --- a/crates/yaak-proxy/tests/own_fallback.rs +++ b/crates/yaak-proxy/tests/own_fallback.rs @@ -7,12 +7,9 @@ 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; @@ -51,11 +48,14 @@ async fn read_head(stream: &mut (impl AsyncReadExt + Unpin)) -> String { 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. +/// A CONNECT is answered only once its server is reached, and on macOS the engine reaches port 80 +/// through Network.framework, which falls back to the system proxy, the engine, when there is no +/// direct route. That fallback is a CONNECT of the engine's own, which dialling again would +/// repeat without end, so it is answered before any dial and refused once it is known to be the +/// engine's own. One `CONNECT` each, then the client gets a 502: for its CONNECT, and for a plain +/// request. #[tokio::test(flavor = "multi_thread")] -async fn own_fallback_through_a_passthrough_host_stops() { +async fn own_fallback_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()); @@ -84,25 +84,16 @@ async fn own_fallback_through_a_passthrough_host_stops() { }); 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 head = tokio::time::timeout(Duration::from_secs(10), read_head(&mut tcp)) + .await + .expect("no answer to the CONNECT"); + 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); 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"); @@ -111,8 +102,6 @@ async fn own_fallback_through_a_passthrough_host_stops() { .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); + assert_eq!(connects.load(Ordering::SeqCst), 2); } diff --git a/crates/yaak-proxy/tests/proxy.rs b/crates/yaak-proxy/tests/proxy.rs index 77b1be8f9..895029544 100644 --- a/crates/yaak-proxy/tests/proxy.rs +++ b/crates/yaak-proxy/tests/proxy.rs @@ -300,16 +300,37 @@ impl Harness { /// A raw connection tunneled through the proxy with CONNECT to `localhost:port`. async fn connect_tunnel(&self, port: u16) -> TcpStream { + self.connect_to(&format!("localhost:{port}")).await + } + + /// A raw connection tunneled through the proxy with CONNECT to `target` (`host:port`). + async fn connect_to(&self, target: &str) -> TcpStream { + let (tcp, head, _) = self.send_connect(target).await; + assert!(head.starts_with("HTTP/1.1 200"), "{head}"); + tcp + } + + /// Sends a CONNECT to `target` and reads the proxy's answer: its head, and its body. + async fn send_connect(&self, target: &str) -> (TcpStream, String, String) { let mut tcp = TcpStream::connect(("127.0.0.1", self.proxy.port)).await.unwrap(); - let connect = - format!("CONNECT localhost:{port} HTTP/1.1\r\nHost: localhost:{port}\r\n\r\n"); + let connect = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n"); tcp.write_all(connect.as_bytes()).await.unwrap(); let mut head = Vec::new(); while !head.ends_with(b"\r\n\r\n") { head.push(tcp.read_u8().await.unwrap()); } - assert!(head.starts_with(b"HTTP/1.1 200"), "{}", String::from_utf8_lossy(&head)); - tcp + let head = String::from_utf8(head).unwrap(); + let length = head + .lines() + .find_map(|line| { + line.to_ascii_lowercase() + .strip_prefix("content-length:") + .map(|n| n.trim().parse::().unwrap()) + }) + .unwrap_or(0); + let mut body = vec![0u8; length]; + tcp.read_exact(&mut body).await.unwrap(); + (tcp, head, String::from_utf8(body).unwrap()) } fn tls_client_config(&self) -> Arc { @@ -407,6 +428,7 @@ fn is_terminal(event: &ProxyEvent) -> bool { | ProxyEvent::StreamEnd { .. } | ProxyEvent::TunnelEnd { .. } | ProxyEvent::HandshakeFailed { .. } + | ProxyEvent::ConnectFailed { .. } ) } @@ -428,7 +450,8 @@ fn flow_id(event: &ProxyEvent) -> u64 { | ProxyEvent::StreamEnd { id, .. } | ProxyEvent::TunnelStart { id, .. } | ProxyEvent::TunnelEnd { id, .. } - | ProxyEvent::HandshakeFailed { id, .. } => *id, + | ProxyEvent::HandshakeFailed { id, .. } + | ProxyEvent::ConnectFailed { id, .. } => *id, } } @@ -1453,21 +1476,204 @@ async fn an_app_entry_decrypts_that_app_alone() { .await } +/// The events of a CONNECT refused for its server: one `ConnectFailed`, and nothing else. +fn connect_failure(h: &Harness) -> (String, String) { + let events = h.flow_events(); + assert_eq!(events.len(), 1, "{events:#?}"); + let ProxyEvent::ConnectFailed { authority, error, .. } = &events[0] else { + panic!("expected the failed connect, got {events:#?}"); + }; + (authority.clone(), error.clone()) +} + +/// A server that can't be reached fails the CONNECT itself, so the client reports a connection +/// error of its own, as it would without the proxy, rather than a response that looks the +/// server's. #[tokio::test(flavor = "multi_thread")] -async fn a_tunnel_to_nowhere_is_a_failed_flow() { +async fn a_name_that_does_not_resolve_fails_the_connect() { + within(async { + let h = Harness::start(1024); + let (_tcp, head, body) = h.send_connect("yaak-proxy-test.invalid:443").await; + assert!(head.starts_with("HTTP/1.1 502"), "{head}"); + assert!( + body.starts_with( + "Yaak Proxy could not connect to yaak-proxy-test.invalid:443: DNS lookup failed: " + ), + "{body}" + ); + let (authority, error) = connect_failure(&h); + assert_eq!(authority, "yaak-proxy-test.invalid"); + assert!(error.starts_with("DNS lookup failed: "), "{error}"); + }) + .await +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_closed_port_fails_the_connect() { + within(async { + // Tunnelled or decrypted, the server is dialled before the CONNECT is answered. + for policy in [&[][..], &["*"][..]] { + let h = Harness::start_with(1024, with_policy(policy)); + let closed = TcpListener::bind("127.0.0.1:0").await.unwrap().local_addr().unwrap(); + let (_tcp, head, body) = h.send_connect(&closed.to_string()).await; + assert!(head.starts_with("HTTP/1.1 502"), "{head}"); + assert!(body.contains(&format!("{closed}: Connection refused: ")), "{body}"); + let (authority, error) = connect_failure(&h); + assert_eq!(authority, closed.to_string()); + assert!(error.starts_with("Connection refused: "), "{error}"); + } + }) + .await +} + +/// A listener on 127.0.0.1 that counts the connections it accepts and echoes what each sends. +async fn spawn_counted_echo() -> (SocketAddr, Arc) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let accepts = Arc::new(AtomicUsize::new(0)); + let counted = accepts.clone(); + tokio::spawn(async move { + loop { + let (mut tcp, _) = listener.accept().await.unwrap(); + counted.fetch_add(1, Ordering::SeqCst); + tokio::spawn(async move { + let (mut read, mut write) = tcp.split(); + let _ = tokio::io::copy(&mut read, &mut write).await; + }); + } + }); + (addr, accepts) +} + +/// Waits until `count` reaches `n`. +async fn wait_for_count(count: &AtomicUsize, n: usize) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while count.load(Ordering::SeqCst) < n { + assert!( + tokio::time::Instant::now() < deadline, + "stayed at {}", + count.load(Ordering::SeqCst) + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_tunnel_relays_over_the_connection_dialled_for_its_connect() { within(async { let h = Harness::start_with(1024, with_policy(&[])); - // A port nothing listens on. - let closed = TcpListener::bind("127.0.0.1:0").await.unwrap().local_addr().unwrap().port(); - let mut tcp = h.connect_tunnel(closed).await; - let _ = tcp.write_all(b"\x16\x03\x01").await; - let mut rest = Vec::new(); - let _ = tcp.read_to_end(&mut rest).await; + let (addr, accepts) = spawn_counted_echo().await; + let mut tcp = h.connect_to(&addr.to_string()).await; + // Connected before the client has sent anything. + wait_for_count(&accepts, 1).await; + tcp.write_all(b"ping").await.unwrap(); + let mut echoed = [0u8; 4]; + tcp.read_exact(&mut echoed).await.unwrap(); + assert_eq!(&echoed, b"ping"); + drop(tcp); let events = h.flow_events(); - let (received, error) = - find!(events, ProxyEvent::TunnelEnd { received, error, .. } => (*received, error.clone())); - assert_eq!(received, 0); - assert!(error.is_some(), "{events:#?}"); + find!(events, ProxyEvent::TunnelStart { .. } => ()); + let error = find!(events, ProxyEvent::TunnelEnd { error, .. } => error.clone()); + assert_eq!(error, None); + assert_eq!(accepts.load(Ordering::SeqCst), 1); + }) + .await +} + +/// `GET /hello` over HTTP/1.1, decrypted, through a CONNECT to `addr`. +async fn decrypted_hello(h: &Harness, addr: SocketAddr) -> String { + let tcp = h.connect_to(&addr.to_string()).await; + let name = ServerName::try_from("localhost").unwrap(); + let tls = TlsConnector::from(h.tls_client_config()).connect(name, tcp).await.unwrap(); + get_hello_and_close(tls).await +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_decrypted_request_goes_on_the_connection_dialled_for_its_connect() { + within(async { + let h = Harness::start(1024); + let accepts = Arc::new(AtomicUsize::new(0)); + let upstream = + spawn_upstream_with(Some(h.pki.server.clone()), accepts.clone(), route).await; + let response = decrypted_hello(&h, upstream).await; + assert!(response.ends_with("hello"), "{response}"); + assert_eq!(accepts.load(Ordering::SeqCst), 1); + + // Its dial is the request's connect time, and the times after it stay in order. + let events = h.flow_events(); + let (connect, tls) = find!(events, + ProxyEvent::UpstreamConnection { elapsed_connect_ms, elapsed_tls_ms, .. } => + (elapsed_connect_ms.unwrap(), elapsed_tls_ms.unwrap())); + let headers_at = find!(events, ProxyEvent::ResponseStart { elapsed_ms, .. } => *elapsed_ms); + let end_at = + find!(events, ProxyEvent::ResponseBodyComplete { elapsed_ms, .. } => *elapsed_ms); + assert!(connect <= tls && tls <= headers_at && headers_at <= end_at, "{events:#?}"); + }) + .await +} + +/// A browser's preconnect that is never used: the server's connection closes with it. +#[tokio::test(flavor = "multi_thread")] +async fn a_client_that_hangs_up_after_the_connect_closes_the_server_connection() { + within(async { + let h = Harness::start(1024); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let tcp = h.connect_to(&addr.to_string()).await; + let (mut server, _) = listener.accept().await.unwrap(); + drop(tcp); + let mut rest = Vec::new(); + let read = tokio::time::timeout(Duration::from_secs(5), server.read_to_end(&mut rest)); + read.await.expect("the server's connection stayed open").unwrap(); + assert!(rest.is_empty()); + }) + .await +} + +/// The server closed the connection dialled for the CONNECT before the first request came, so +/// the request goes on a new one. +#[tokio::test(flavor = "multi_thread")] +async fn a_dialled_connection_the_server_closed_is_dialled_again() { + within(async { + let h = Harness::start(1024); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let accepts = Arc::new(AtomicUsize::new(0)); + let counted = accepts.clone(); + let tls = TlsAcceptor::from(h.pki.server.clone()); + tokio::spawn(async move { + loop { + let (tcp, _) = listener.accept().await.unwrap(); + // The first is closed unused, as an idle server would. + if counted.fetch_add(1, Ordering::SeqCst) == 0 { + drop(tcp); + continue; + } + let tls = tls.clone(); + tokio::spawn(async move { + let Ok(stream) = tls.accept(tcp).await else { + return; + }; + let builder = auto::Builder::new(TokioExecutor::new()); + let _ = builder.serve_connection(TokioIo::new(stream), service_fn(route)).await; + }); + } + }); + + let tcp = h.connect_to(&addr.to_string()).await; + wait_for_count(&accepts, 1).await; + let name = ServerName::try_from("localhost").unwrap(); + let mut tls = TlsConnector::from(h.tls_client_config()).connect(name, tcp).await.unwrap(); + // Long enough for the close to arrive before the request. + tokio::time::sleep(Duration::from_millis(100)).await; + tls.write_all(b"GET /hello HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + let mut response = Vec::new(); + let _ = tls.read_to_end(&mut response).await; + let response = String::from_utf8_lossy(&response); + assert!(response.starts_with("HTTP/1.1 200") && response.ends_with("hello"), "{response}"); + assert_eq!(accepts.load(Ordering::SeqCst), 2); }) .await }