openlatch-client 0.6.19

//! `CONNECT host:port` from an agent pointed at the relay through its proxy setting (I-2 plan
//! 01, §3). Answer 200, then either terminate TLS inside the upgraded tunnel and serve
//! HTTP/1.1 into the same router, or copy bytes to an opaque tunnel opened through the egress
//! route. Never a refusal on the host: an undeclared host is the normal case (the agent's
//! auth, telemetry and updates), not an error.

use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use axum::body::Body;
use axum::extract::Request;
use axum::http::StatusCode;
use axum::response::Response;
use hyper::upgrade::Upgraded;
use hyper_util::rt::TokioIo;

use super::{proxy, ModelRelayState};
use crate::error::OlError;

/// Request extension the inner service sets on every request that arrived inside an
/// intercepted tunnel. `proxy::upstream_url` reads it. Absent on the redirect path.
#[derive(Clone, Debug)]
pub(crate) struct ConnectAuthority {
    pub(crate) host: String,
    pub(crate) port: u16,
    /// Whether this tunnel's host was flagged h2-capable when the tunnel opened (Cursor I-2
    /// D-01) — decided once, from the same config that chose its ALPN list, so every request on
    /// the tunnel is judged alike and none takes a lock to find out.
    pub(crate) h2: bool,
}

const INTERCEPT_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);

/// What `handle_connect` decided before it answers `200` and returns: either the CONNECT
/// authority is declared and gets TLS terminated in-process, or it is opaque and gets a plain
/// byte tunnel through the egress route.
enum Plan {
    /// The leaf config chosen at CONNECT, kept for the handshake: a pattern host's leaf evicted
    /// from the LRU (or re-minted) while the ClientHello is awaited never closes a CONNECT already
    /// answered `200` (AGT-6 plan D-9).
    Intercept(Arc<rustls::ServerConfig>),
    Tunnel(crate::egress::tunnel::TunnelStream),
}

/// `CONNECT host:port`, on the relay's main listener only (`proxy_any` already checked
/// `st.endpoint.is_none()`). Answers `200` immediately and finishes the work after hyper
/// completes the upgrade — hyper's `examples/http_proxy.rs` shape, with one deliberate
/// difference: an opaque tunnel is OPENED BEFORE the `200`, so a route failure is a `502` the
/// client sees, never a `200` then a dead socket.
pub async fn handle_connect(st: Arc<ModelRelayState>, mut req: Request<Body>) -> Response {
    let Some(authority) = req.uri().authority().cloned() else {
        let mut resp = Response::new(Body::empty());
        *resp.status_mut() = StatusCode::BAD_REQUEST;
        return resp;
    };
    // 01-20: IPv6 is BARE ("::1") for the socket connect, the no_proxy match and minting (an IP
    // SAN); brackets come back only where an authority string is written (§2 step 4, §5's
    // `https://{host}:{port}` URL, and the PAC target URL).
    let mut host = authority
        .host()
        .trim_start_matches('[')
        .trim_end_matches(']')
        .to_ascii_lowercase();
    let port = authority.port_u16().unwrap_or(443);
    // Our own probe (plan 02's `probe_intercept`) marks its CONNECT head with this header; its
    // handshake must never reset an agent's D-32 count.
    let preflight = proxy::is_preflight(req.headers());

    // Clone the Arc and drop the read guard before anything else runs — never held across an
    // await (the "Lock order" note in plan 01 §1).
    let slot = st
        .intercept
        .read()
        .unwrap_or_else(|e| e.into_inner())
        .clone();
    // D-02 is host-only: no `port == 443` test. The leaf carries the host, not a port, and the
    // forward leg (`upstream_url`) keeps the authority's port.
    //
    // AGT-6 C-5 / D-9: a declared host is an exact host or a pattern match, and a pattern host's
    // leaf is minted HERE, at CONNECT, outside every lock (`ensure_leaf`). Past the per-minute
    // mint cap, or if the mint fails, the CONNECT is tunneled untouched like any undeclared host.
    let plan = match slot.and_then(|ic| ic.connect_config(&host)) {
        // The intercepted tunnel works in the matcher's shape (one trailing dot dropped), so the
        // SNI check and the forward leg compare like with like. A tunnel keeps the raw authority.
        Some(config) => {
            host = super::host_match::normalize(&host).into_owned();
            Plan::Intercept(config)
        }
        None => match open_upstream(&st, &host, port).await {
            Ok(up) => Plan::Tunnel(up),
            Err(e) => {
                tracing::warn!(
                    code = %e.code,
                    host = %host,
                    port,
                    "CONNECT tunnel could not be opened"
                );
                // Bumps the process-wide `upstream_failures()` counter the wiring supervisor
                // re-probes on: "the agent got nothing back", and a re-probe is cheap.
                return proxy::synth_502(&st);
            }
        },
    };

    // BEFORE the response leaves this function — hyper completes the upgrade only after the
    // `200` has been sent.
    let upgrade = hyper::upgrade::on(&mut req);
    tokio::spawn(async move {
        let upgraded = match upgrade.await {
            Ok(u) => u,
            Err(e) => {
                tracing::debug!(error = %e, "model relay CONNECT upgrade failed");
                return;
            }
        };
        let io = TokioIo::new(upgraded);
        match plan {
            Plan::Intercept(config) => terminate(st, config, io, host, port, preflight).await,
            Plan::Tunnel(mut upstream) => {
                let mut io = io;
                let _ = tokio::io::copy_bidirectional(&mut io, &mut upstream).await;
            }
        }
    });

    Response::new(Body::empty())
}

/// The tunnel's far end for an undeclared (or slot-empty) CONNECT authority: the test seam
/// first, then the real egress route. `st.egress` is the same `EgressReporter` the forward
/// client was built from — the two legs share one route with no new field.
async fn open_upstream(
    st: &Arc<ModelRelayState>,
    host: &str,
    port: u16,
) -> Result<crate::egress::tunnel::TunnelStream, OlError> {
    // Empty in production (`with_resolve_override` is a test seam): skip the lookup key
    // allocation on every CONNECT rather than build and hash a `(String, u16)` for a map
    // nothing ever populated.
    if !st.resolve_override.is_empty() {
        if let Some(addr) = st.resolve_override.get(&(host.to_string(), port)) {
            return crate::egress::tunnel::connect_with_timeout(*addr, host, port).await;
        }
    }
    crate::egress::tunnel::open(st.egress.config(), host, port).await
}

/// Terminate TLS inside the upgraded tunnel for a DECLARED host, then serve HTTP/1.1 on the
/// decrypted stream into the same router the main listener uses.
///
/// A server handshake completes only on the client's Finished, which a client that rejects our
/// certificate never sends — so a refusal always surfaces HERE, never after a success.
async fn terminate(
    st: Arc<ModelRelayState>,
    pinned: Arc<rustls::ServerConfig>,
    io: TokioIo<Upgraded>,
    host: String,
    port: u16,
    preflight: bool,
) {
    // Read the ClientHello WITHOUT committing to a config.
    let start = match tokio::time::timeout(
        INTERCEPT_HANDSHAKE_TIMEOUT,
        tokio_rustls::LazyConfigAcceptor::new(rustls::server::Acceptor::default(), io),
    )
    .await
    {
        Ok(Ok(start)) => start,
        // Timeout or a closed socket: the client never saw our leaf. NOT a refusal.
        _ => {
            tracing::debug!(host = %host, "model relay intercept closed before a ClientHello");
            return;
        }
    };

    // D-02, decided here and not counted: the pinned resolver would refuse a mismatched SNI
    // too, but deciding it HERE is what keeps our own refusal out of the D-32 count without
    // matching rustls's error string.
    let sni_ok = start
        .client_hello()
        .server_name()
        .is_none_or(|s| s.eq_ignore_ascii_case(&host));
    let config = sni_ok.then_some(pinned);
    let Some(config) = config else {
        tracing::debug!(
            host = %host,
            "model relay intercept SNI does not match the CONNECT authority"
        );
        return;
    };
    // Cursor I-2 D-01: only an h2-flagged host's config offers `h2` (`ca::server_config_with`).
    let h2 = config.alpn_protocols.iter().any(|p| p == b"h2");

    // Offer the leaf. From here a failure is the PEER refusing it (D-32). A preflight probe's
    // three outcome arms below record NOTHING on `st.refusals` (neither success nor refusal);
    // the pass-through-failure counter and the return are unchanged.
    let tls =
        match tokio::time::timeout(INTERCEPT_HANDSHAKE_TIMEOUT, start.into_stream(config)).await {
            Ok(Ok(tls)) => {
                if !preflight {
                    st.refusals.record_success(&host);
                }
                tls
            }
            Ok(Err(e)) => {
                if !preflight {
                    st.refusals.record_refusal(&host, classify_refusal(&e));
                }
                proxy::record_pass_through_failure("intercept_handshake");
                return;
            }
            Err(_) => {
                if !preflight {
                    st.refusals
                        .record_refusal(&host, RefusalKind::Other("handshake timed out".into()));
                }
                proxy::record_pass_through_failure("intercept_handshake");
                return;
            }
        };

    // Serve HTTP/1.1 on the decrypted stream INTO THE SAME ROUTER. `tower` is not a dependency;
    // axum's own `Extension` layer inserts the value into every request reaching this router.
    let svc = super::router(st).layer(axum::Extension(ConnectAuthority { host, port, h2 }));
    // Cursor I-2 D-02: ALPN lives on the connection half, and is read BEFORE the stream moves
    // into `TokioIo`. Only an h2-flagged host's leaf ever offers `h2` (`ca::server_config_with`),
    // so every other tunnel takes the HTTP/1.1 path below, unchanged.
    if tls.get_ref().1.alpn_protocol() == Some(b"h2") {
        serve_h2(tls, svc).await;
        return;
    }
    if let Err(e) = hyper::server::conn::http1::Builder::new()
        .serve_connection(
            TokioIo::new(tls),
            hyper_util::service::TowerToHyperService::new(svc),
        )
        .with_upgrades()
        .await
    {
        // A client hanging up mid-stream is normal — never a refusal, never a 502.
        tracing::debug!(error = %e, "model relay intercept connection ended");
    }
    // A CONNECT that arrives INSIDE a tunnel reaches `proxy_any` again through this same
    // router and is handled like any other — it cannot loop (each hop is a new upgrade on the
    // agent's own connection).
}

/// HTTP/2 on a decrypted tunnel that negotiated `h2` (Cursor I-2 D-02): the SAME router and
/// `ConnectAuthority` layer as the HTTP/1.1 path, so a request reaches `proxy_any` exactly as an
/// HTTP/1.1 one does. Every stream is its own task on the connection, so a long bidirectional
/// stream never blocks a second request on the same connection.
async fn serve_h2(tls: tokio_rustls::server::TlsStream<TokioIo<Upgraded>>, svc: axum::Router) {
    if let Err(e) = hyper::server::conn::http2::Builder::new(hyper_util::rt::TokioExecutor::new())
        .serve_connection(
            TokioIo::new(tls),
            hyper_util::service::TowerToHyperService::new(svc),
        )
        .await
    {
        // As on the HTTP/1.1 path: a client hanging up mid-stream is normal.
        tracing::debug!(error = %e, "model relay intercept h2 connection ended");
    }
}

/// How a peer refused our leaf. The signatures the POC observed (PR #463,
/// `tools/mitm-relay-poc/proof/README.md`): VS Code's Node path closes mid-handshake with no
/// alert (tokio-rustls: `UnexpectedEof`, "tls handshake eof"); Bun resets the connection;
/// Chromium and rustls send `UnknownCA`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RefusalKind {
    Alert(String),
    HandshakeEof,
    Reset,
    Other(String),
}

/// Classify a server-side handshake failure into a [`RefusalKind`]. Total, never panics.
fn classify_refusal(e: &std::io::Error) -> RefusalKind {
    if let Some(rustls::Error::AlertReceived(d)) =
        e.get_ref().and_then(|i| i.downcast_ref::<rustls::Error>())
    {
        return RefusalKind::Alert(format!("{d:?}"));
    }
    match e.kind() {
        std::io::ErrorKind::UnexpectedEof => RefusalKind::HandshakeEof,
        std::io::ErrorKind::ConnectionReset
        | std::io::ErrorKind::ConnectionAborted
        | std::io::ErrorKind::BrokenPipe => RefusalKind::Reset,
        _ => RefusalKind::Other(e.to_string()),
    }
}

/// A host trips after this many consecutive refusals with no success in [`SUCCESS_WINDOW`].
const TRIP_AFTER: u32 = 3;
const SUCCESS_WINDOW: Duration = Duration::from_secs(10 * 60);

#[derive(Default)]
struct HostRecord {
    consecutive: u32,
    last_success: Option<Instant>,
    tripped_logged: bool,
}

/// D-32's counting half. Plan 04 reads [`RefusalLedger::tripped_hosts`] and reacts; this
/// module never releases, renders or unwires anything.
#[derive(Default)]
pub struct RefusalLedger {
    hosts: Mutex<BTreeMap<String, HostRecord>>,
}

impl RefusalLedger {
    /// Record a refusal for `host` (case-insensitive). Warns once when the host first trips.
    pub fn record_refusal(&self, host: &str, kind: RefusalKind) {
        let host = host.to_ascii_lowercase();
        let mut hosts = self.hosts.lock().unwrap_or_else(|e| e.into_inner());
        let record = hosts.entry(host.clone()).or_default();
        record.consecutive += 1;
        tracing::debug!(host = %host, kind = ?kind, "model relay intercept handshake refused");
        if record.consecutive >= TRIP_AFTER && !record.tripped_logged {
            record.tripped_logged = true;
            tracing::warn!(host = %host, kind = ?kind, "model relay intercept host tripped");
        }
    }

    /// Record a completed handshake for `host` (case-insensitive), resetting its count.
    pub fn record_success(&self, host: &str) {
        self.record_success_at(host, Instant::now());
    }

    /// Every host with `>= TRIP_AFTER` consecutive refusals and no success within
    /// [`SUCCESS_WINDOW`], sorted.
    pub fn tripped_hosts(&self) -> Vec<String> {
        self.tripped_hosts_at(Instant::now())
    }

    /// Forget `host` entirely. A tripped host is re-armed by a daemon restart (a fresh
    /// `RefusalLedger`), or by `daemon::react_to_trust_loss_in` calling this once the agent's
    /// TLS-trust process was replaced (a Cursor window reload).
    pub fn clear(&self, host: &str) {
        let host = host.to_ascii_lowercase();
        self.hosts
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(&host);
    }

    fn record_success_at(&self, host: &str, at: Instant) {
        let host = host.to_ascii_lowercase();
        let mut hosts = self.hosts.lock().unwrap_or_else(|e| e.into_inner());
        let record = hosts.entry(host).or_default();
        record.consecutive = 0;
        record.last_success = Some(at);
        record.tripped_logged = false;
    }

    fn tripped_hosts_at(&self, now: Instant) -> Vec<String> {
        let hosts = self.hosts.lock().unwrap_or_else(|e| e.into_inner());
        hosts
            .iter()
            .filter(|(_, r)| {
                r.consecutive >= TRIP_AFTER
                    && r.last_success
                        .is_none_or(|t| now.saturating_duration_since(t) > SUCCESS_WINDOW)
            })
            .map(|(h, _)| h.clone())
            .collect()
    }
}

#[cfg(test)]
mod tests {
    use super::super::ca;
    use super::*;
    use rustls::pki_types::pem::PemObject;

    #[test]
    fn a_tls_alert_is_classified_by_its_description() {
        let err = std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            rustls::Error::AlertReceived(rustls::AlertDescription::UnknownCA),
        );
        assert_eq!(
            classify_refusal(&err),
            RefusalKind::Alert("UnknownCA".to_string())
        );
    }

    #[test]
    fn eof_reset_and_other_are_classified() {
        let eof = std::io::Error::from(std::io::ErrorKind::UnexpectedEof);
        assert_eq!(classify_refusal(&eof), RefusalKind::HandshakeEof);

        for kind in [
            std::io::ErrorKind::ConnectionReset,
            std::io::ErrorKind::ConnectionAborted,
            std::io::ErrorKind::BrokenPipe,
        ] {
            assert_eq!(
                classify_refusal(&std::io::Error::from(kind)),
                RefusalKind::Reset
            );
        }

        let other = std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            rustls::Error::NoApplicationProtocol,
        );
        assert!(matches!(classify_refusal(&other), RefusalKind::Other(_)));
    }

    #[test]
    fn three_consecutive_refusals_trip_a_host() {
        let ledger = RefusalLedger::default();
        ledger.record_refusal("a.test", RefusalKind::Reset);
        ledger.record_refusal("a.test", RefusalKind::Reset);
        assert!(ledger.tripped_hosts().is_empty());
        ledger.record_refusal("a.test", RefusalKind::Reset);
        assert_eq!(ledger.tripped_hosts(), vec!["a.test".to_string()]);
        ledger.record_refusal("b.test", RefusalKind::Reset);
        assert_eq!(ledger.tripped_hosts(), vec!["a.test".to_string()]);
    }

    #[test]
    fn a_success_within_ten_minutes_holds_the_trip() {
        let ledger = RefusalLedger::default();
        let t0 = Instant::now();
        ledger.record_success_at("a.test", t0);
        ledger.record_refusal("a.test", RefusalKind::Reset);
        ledger.record_refusal("a.test", RefusalKind::Reset);
        ledger.record_refusal("a.test", RefusalKind::Reset);
        assert!(ledger
            .tripped_hosts_at(t0 + Duration::from_secs(9 * 60 + 59))
            .is_empty());
        assert_eq!(
            ledger.tripped_hosts_at(t0 + Duration::from_secs(10 * 60 + 1)),
            vec!["a.test".to_string()]
        );
    }

    #[test]
    fn a_success_resets_the_count() {
        let ledger = RefusalLedger::default();
        ledger.record_refusal("a.test", RefusalKind::Reset);
        ledger.record_refusal("a.test", RefusalKind::Reset);
        ledger.record_success("a.test");
        ledger.record_refusal("a.test", RefusalKind::Reset);
        ledger.record_refusal("a.test", RefusalKind::Reset);
        assert!(ledger
            .tripped_hosts_at(Instant::now() + Duration::from_secs(11 * 60))
            .is_empty());
    }

    #[test]
    fn clear_forgets_a_host() {
        let ledger = RefusalLedger::default();
        ledger.record_refusal("a.test", RefusalKind::Reset);
        ledger.record_refusal("a.test", RefusalKind::Reset);
        ledger.record_refusal("a.test", RefusalKind::Reset);
        assert_eq!(ledger.tripped_hosts(), vec!["a.test".to_string()]);
        ledger.clear("a.test");
        assert!(ledger.tripped_hosts().is_empty());
        ledger.record_refusal("a.test", RefusalKind::Reset);
        assert!(ledger.tripped_hosts().is_empty());
    }

    /// A raw CONNECT carrying our own preflight header, against a DECLARED host whose
    /// forward leg is pinned to a closed port (hermetic 502 on the inner request): three
    /// refusals trip the host directly; the preflight's own successful handshake — proven by
    /// reading a response, any status, over the decrypted stream — must not clear that trip.
    /// The control (no header) resets the count.
    #[tokio::test]
    async fn a_preflight_connect_is_not_counted() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let interceptor = Arc::new(
            ca::Interceptor::new(&tmp.path().join("ca"), ["a.test"], None::<&str>).expect("new"),
        );
        let ledger = Arc::new(RefusalLedger::default());

        let closed = crate::model_relay::mock::closed_port().await;
        let mut resolve = BTreeMap::new();
        resolve.insert(
            ("a.test".to_string(), 443u16),
            std::net::SocketAddr::from(([127, 0, 0, 1], closed)),
        );

        let slot: super::super::InterceptSlot =
            Arc::new(std::sync::RwLock::new(Some(interceptor.clone())));
        let state = Arc::new(
            ModelRelayState::new(
                reqwest::Url::parse("http://127.0.0.1:1").unwrap(),
                0,
                8,
                &[],
            )
            .with_intercept(slot)
            .with_refusals(ledger.clone())
            .with_resolve_override(resolve),
        );
        let port = super::super::serve_ephemeral(state).await;

        // Trip the host directly, then prove a preflight CONNECT's own successful
        // handshake — proven by reading a response over the decrypted stream — leaves the
        // trip alone.
        ledger.record_refusal("a.test", RefusalKind::Reset);
        ledger.record_refusal("a.test", RefusalKind::Reset);
        ledger.record_refusal("a.test", RefusalKind::Reset);
        assert_eq!(ledger.tripped_hosts(), vec!["a.test".to_string()]);
        probe(port, &interceptor.ca().pem_path(), true).await;
        assert_eq!(
            ledger.tripped_hosts(),
            vec!["a.test".to_string()],
            "a preflight connect's handshake must not be recorded, success or refusal"
        );

        // Control: the SAME connect with no header is a real success and resets the count.
        probe(port, &interceptor.ca().pem_path(), false).await;
        assert!(ledger.tripped_hosts().is_empty());
    }

    /// The status line of a raw `CONNECT <host>:443` to the relay at `port`, and the stream.
    async fn connect_head(port: u16, host: &str) -> (String, tokio::net::TcpStream) {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port))
            .await
            .expect("dial relay");
        stream
            .write_all(
                format!("CONNECT {host}:443 HTTP/1.1\r\nHost: {host}:443\r\n\r\n").as_bytes(),
            )
            .await
            .expect("write CONNECT");
        let mut buf = Vec::new();
        let mut byte = [0u8; 1];
        while !buf.ends_with(b"\r\n\r\n") {
            stream
                .read_exact(&mut byte)
                .await
                .expect("read CONNECT response");
            buf.push(byte[0]);
        }
        let head = String::from_utf8_lossy(&buf);
        (head.lines().next().unwrap_or_default().to_string(), stream)
    }

    /// AGT-6 C-5 / D-9, through the real CONNECT path: a pattern host gets its own leaf minted at
    /// CONNECT and a TLS handshake trusting our CA completes against it; once eight leaves were
    /// minted in the minute, the next pattern host is tunneled untouched (its far end sees the
    /// client's own bytes) and counted. An undeclared host is tunneled and mints nothing.
    #[tokio::test]
    async fn pattern_hosts_mint_at_connect_until_the_cap_then_tunnel() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let tmp = tempfile::tempdir().expect("tempdir");
        let interceptor = Arc::new(
            ca::Interceptor::new(&tmp.path().join("ca"), ["a.test"], None::<&str>).expect("new"),
        );
        interceptor.set_patterns([super::super::host_match::HostRule::OneLabel {
            prefix: "",
            suffix: ".pat.test",
        }]);

        // A plain TCP far end for the tunneled hosts: it reports the first bytes it receives.
        let far = tokio::net::TcpListener::bind(("127.0.0.1", 0))
            .await
            .expect("bind far end");
        let far_addr = far.local_addr().expect("far addr");
        let (seen_tx, mut seen_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
        tokio::spawn(async move {
            while let Ok((mut sock, _)) = far.accept().await {
                let tx = seen_tx.clone();
                tokio::spawn(async move {
                    let mut buf = [0u8; 5];
                    if sock.read_exact(&mut buf).await.is_ok() {
                        let _ = tx.send(buf.to_vec());
                    }
                });
            }
        });

        let closed = crate::model_relay::mock::closed_port().await;
        let mut resolve = BTreeMap::new();
        for i in 0..8 {
            resolve.insert(
                (format!("h{i}.pat.test"), 443u16),
                std::net::SocketAddr::from(([127, 0, 0, 1], closed)),
            );
        }
        resolve.insert(("over.pat.test".to_string(), 443u16), far_addr);
        resolve.insert(("z.test".to_string(), 443u16), far_addr);

        let slot: super::super::InterceptSlot =
            Arc::new(std::sync::RwLock::new(Some(interceptor.clone())));
        let state = Arc::new(
            ModelRelayState::new(
                reqwest::Url::parse("http://127.0.0.1:1").unwrap(),
                0,
                8,
                &[],
            )
            .with_intercept(slot)
            .with_resolve_override(resolve),
        );
        let port = super::super::serve_ephemeral(state).await;

        let ca_pem = std::fs::read_to_string(interceptor.ca().pem_path()).expect("read ca pem");
        let ca_der =
            rustls::pki_types::CertificateDer::from_pem_slice(ca_pem.as_bytes()).expect("parse");
        let mut roots = rustls::RootCertStore::empty();
        roots.add(ca_der).expect("trust the CA");
        let connector = tokio_rustls::TlsConnector::from(Arc::new(
            rustls::ClientConfig::builder()
                .with_root_certificates(roots)
                .with_no_client_auth(),
        ));

        for i in 0..8 {
            let host = format!("h{i}.pat.test");
            let (status, stream) = connect_head(port, &host).await;
            assert!(status.starts_with("HTTP/1.1 200"), "{host}: {status}");
            let name = rustls::pki_types::ServerName::try_from(host.clone()).expect("name");
            connector
                .connect(name, stream)
                .await
                .unwrap_or_else(|e| panic!("{host}: handshake against its own leaf: {e}"));
        }
        assert_eq!(interceptor.pattern_mint_capped(), 0);

        // The ninth pattern host in the minute: tunneled, its bytes reach the far end untouched.
        let (status, mut stream) = connect_head(port, "over.pat.test").await;
        assert!(status.starts_with("HTTP/1.1 200"), "{status}");
        stream
            .write_all(b"hello")
            .await
            .expect("write through tunnel");
        let seen = tokio::time::timeout(Duration::from_secs(5), seen_rx.recv())
            .await
            .expect("far end saw the tunneled bytes")
            .expect("channel open");
        assert_eq!(seen, b"hello");
        assert_eq!(interceptor.pattern_mint_capped(), 1);
        assert!(interceptor.pinned_config("over.pat.test").is_none());

        // An undeclared host: tunneled, and not counted as a capped mint.
        let (status, mut stream) = connect_head(port, "z.test").await;
        assert!(status.starts_with("HTTP/1.1 200"), "{status}");
        stream
            .write_all(b"plain")
            .await
            .expect("write through tunnel");
        let seen = tokio::time::timeout(Duration::from_secs(5), seen_rx.recv())
            .await
            .expect("far end saw the tunneled bytes")
            .expect("channel open");
        assert_eq!(seen, b"plain");
        assert_eq!(interceptor.pattern_mint_capped(), 1);
    }

    /// `CONNECT a.test:443` to the relay at `port`, then a completed TLS handshake trusting
    /// `ca_pem_path`, then one `GET /` whose response (any status) is read to the end — so the
    /// server's handshake-success arm has certainly run before this returns.
    async fn probe(port: u16, ca_pem_path: &std::path::Path, preflight: bool) {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port))
            .await
            .expect("dial relay");
        let mut head = "CONNECT a.test:443 HTTP/1.1\r\nHost: a.test:443\r\n".to_string();
        if preflight {
            head.push_str(&format!(
                "{}: 1\r\n",
                super::super::preflight::PREFLIGHT_HEADER
            ));
        }
        head.push_str("\r\n");
        stream
            .write_all(head.as_bytes())
            .await
            .expect("write CONNECT");

        let mut buf = Vec::new();
        let mut byte = [0u8; 1];
        while !buf.ends_with(b"\r\n\r\n") {
            stream
                .read_exact(&mut byte)
                .await
                .expect("read CONNECT response");
            buf.push(byte[0]);
        }
        assert!(
            String::from_utf8_lossy(&buf).starts_with("HTTP/1.1 200"),
            "unexpected CONNECT response: {}",
            String::from_utf8_lossy(&buf)
        );

        let ca_pem = std::fs::read_to_string(ca_pem_path).expect("read ca pem");
        let ca_der =
            rustls::pki_types::CertificateDer::from_pem_slice(ca_pem.as_bytes()).expect("parse");
        let mut roots = rustls::RootCertStore::empty();
        roots.add(ca_der).expect("trust the CA");
        let client_config = rustls::ClientConfig::builder()
            .with_root_certificates(roots)
            .with_no_client_auth();
        let connector = tokio_rustls::TlsConnector::from(Arc::new(client_config));
        let server_name = rustls::pki_types::ServerName::try_from("a.test").expect("server name");
        let mut tls = connector
            .connect(server_name, stream)
            .await
            .expect("tls handshake against our own leaf must succeed");

        tls.write_all(b"GET / HTTP/1.1\r\nHost: a.test\r\nConnection: close\r\n\r\n")
            .await
            .expect("write GET");
        let mut response = Vec::new();
        let _ = tls.read_to_end(&mut response).await;
    }
}