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 <noreply@anthropic.com>

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,
+    },
   ];
 }
 

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 [
         {

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::<Utc>::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);

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()
     }

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 {

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<Value> {
+        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.

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
     };

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<SocketAddr> {
+        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<Dialled, Unreachable> {
+    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<UpstreamIo, Unreachable> {
+    #[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<SocketAddr> = 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<io::Result<()>> {
+        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<io::Result<usize>> {
+        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<io::Result<usize>> {
+        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<io::Result<()>> {
+        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<io::Result<()>> {
+        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}");
+    }
+}

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<u64>,
         elapsed_tls_ms: Option<u64>,
         /// `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.

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::<LookupFailed>())
+}
+
 /// 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<NwStream> {
     open(host, port, |_| {}).await

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<Engine>,
 ) -> Result<Response<BoxBody>, 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<Incoming>,
     client: ClientConn,
     engine: Arc<Engine>,
@@ -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<BoxBody> {
+    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<S>(
     mut client: S,
     target: Target,
+    dialled: Option<Dialled>,
     conn: ClientConn,
     engine: Arc<Engine>,
 ) -> 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<S>(
+    client: S,
+    target: &Target,
+    upstream: Option<UpstreamIo>,
+    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<S>(
     client: S,
     target: &Target,
+    upstream: Option<UpstreamIo>,
     conn: ClientConn,
     engine: &Engine,
     reason: TunnelReason,
@@ -531,7 +612,7 @@ async fn tunnel_flow<S>(
 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<Duration>,
     handshaken: Instant,
     version: Option<String>,
     cipher: Option<String>,
 }
 
 impl TlsInfo {
-    fn of(conn: &rustls::ClientConnection, connected: Instant) -> Self {
+    fn of(
+        conn: &rustls::ClientConnection,
+        connected: Instant,
+        dialled_before: Option<Duration>,
+    ) -> 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<Option<Dialled>>,
     conns: tokio::sync::Mutex<Vec<UpstreamConn>>,
     /// 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<Dialled>) -> 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<UpstreamConn, BoxError> {
+        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<Dialled>,
 ) -> Result<UpstreamConn, BoxError> {
-    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<OnUpgrade> =
         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()))),
         };

crates/yaak-proxy/src/tunnel.rs

@@ -108,7 +108,8 @@ where
     relay(client, crate::nw::connect(host, port).await?, prefix).await
 }
 
-async fn relay<S, U>(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<S, U>(mut client: S, mut upstream: U, prefix: &[u8]) -> io::Result<()>
 where
     S: AsyncRead + AsyncWrite + Unpin,
     U: AsyncRead + AsyncWrite + Unpin,

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);
 }

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::<usize>().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<ClientConfig> {
@@ -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<AtomicUsize>) {
+    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
 }