openlatch-client 0.6.1

//! An opaque TCP tunnel to `host:port`, opened the way the model relay's forward client reaches a
//! provider, so a CONNECT the relay does not intercept still leaves a proxied estate through the
//! corporate proxy. Mirrors `factory::build_proxy`'s route decision step for step, then speaks the
//! CONNECT that `resolve::read_head` already reads one byte at a time.
//!
//! I-2 §2. Nothing here opens a socket to a remote host outside this module's two paths
//! (direct, or CONNECT through `[proxy]`) — see AGENTS.md, "nothing leaves the host outside
//! `core::egress`".

use std::borrow::Cow;
use std::time::Duration;

use base64::Engine;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;

use crate::core::error::{
    OlError, ERR_EGRESS_UNREACHABLE, ERR_PROXY_AUTH_FAILED, ERR_PROXY_SCHEME_UNSUPPORTED,
    ERR_PROXY_UNREACHABLE,
};

use super::config::{EgressConfig, ProxyAuth, ProxyMode};
use super::discovery;
use super::factory;
use super::resolve::{parse_connect_status, read_head};

/// Both routes end in a plain TCP stream today. An alias rather than an enum, so the contract
/// name exists without a wrapper type that only forwards `AsyncRead`/`AsyncWrite`.
pub type TunnelStream = TcpStream;

/// How long a direct or proxy-hop TCP connect is allowed to take before the CONNECT that
/// opened it is answered with a synthetic 502.
pub const TUNNEL_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
/// How long the proxy's CONNECT response head is allowed to take.
const TUNNEL_HEAD_TIMEOUT: Duration = Duration::from_secs(10);

/// Decided BEFORE any agent is wired, without opening a socket. Plan 02's `probe_intercept`
/// calls this first, so on an estate whose route cannot carry an opaque tunnel no `ProxyEnv`
/// agent is ever pointed at the relay, and doctor renders the reason (plan 04). Returns the
/// sentence doctor shows.
pub fn route_supported(cfg: &EgressConfig) -> Result<(), String> {
    if cfg.mode == ProxyMode::Direct {
        return Ok(());
    }
    // A PAC-sourced configuration answers per destination through the OS evaluator, so
    // whether the estate can carry a tunnel is not knowable ahead of time — except for the
    // one thing `[proxy] allow_direct = false` forbids: a PAC that may legally answer DIRECT,
    // or fail to evaluate at all, leaving the request unproxied.
    if discovery::pac_binding(cfg).is_some() {
        if !cfg.allow_direct {
            return Err(factory::PAC_DIRECT_FORBIDDEN.to_string());
        }
        return Ok(());
    }
    let route = match factory::effective_route(cfg) {
        Ok(route) => route,
        Err(e) => return Err(e.message),
    };
    let Some(url) = route else {
        return Ok(());
    };
    let Ok(parsed) = reqwest::Url::parse(&url) else {
        // Not reachable today: `effective_route` only ever returns a URL this crate itself
        // wrote (from `[proxy] url` or the standard environment), both already validated at
        // parse time. Refuse rather than silently treat an unparseable route as unsupported.
        return Err(format!(
            "the configured proxy route \"{url}\" could not be parsed"
        ));
    };
    match parsed.scheme() {
        "https" => Err("an https:// proxy cannot carry the relay's opaque tunnels yet".to_string()),
        "socks5" | "socks5h" => {
            Err("a SOCKS proxy cannot carry the relay's opaque tunnels yet".to_string())
        }
        _ if cfg.auth == ProxyAuth::Negotiate => Err(
            "a Negotiate (Kerberos) proxy cannot carry the relay's opaque tunnels yet".to_string(),
        ),
        _ => Ok(()),
    }
}

/// Bracket a bare IPv6 host for an authority string ("host:port" becomes "[host]:port"); an
/// IPv4 literal or a DNS name passes through unchanged.
///
/// 01-20: every host this crate carries internally — a CONNECT's parsed authority, the
/// resolve seam's key — is BARE, no brackets, because a bracketed string is not a valid
/// socket-connect host and is not what an IP SAN or a `no_proxy` match wants. Brackets exist
/// only where an authority is written back out to the wire or into a URL: the proxy CONNECT
/// line, an upstream URL, a PAC target URL.
pub(crate) fn bracket_host(host: &str) -> Cow<'_, str> {
    if host.contains(':') {
        Cow::Owned(format!("[{host}]"))
    } else {
        Cow::Borrowed(host)
    }
}

/// `TcpStream::connect(addr)` under [`TUNNEL_CONNECT_TIMEOUT`], both failure arms mapped to
/// [`ERR_EGRESS_UNREACHABLE`] naming `host:port` (the address the caller resolved `addr`
/// from, kept separate because `addr` itself may be a pinned test address rather than
/// `host:port`'s own resolution). Shared by [`connect_direct`] and `intercept::open_upstream`'s
/// resolve-override test seam — the same connect-and-classify, a proxy route or a pinned
/// loopback address either one.
pub(crate) async fn connect_with_timeout<A: tokio::net::ToSocketAddrs>(
    addr: A,
    host: &str,
    port: u16,
) -> Result<TunnelStream, OlError> {
    match tokio::time::timeout(TUNNEL_CONNECT_TIMEOUT, TcpStream::connect(addr)).await {
        Ok(Ok(stream)) => Ok(stream),
        Ok(Err(e)) => Err(OlError::new(
            ERR_EGRESS_UNREACHABLE,
            format!("could not reach {host}:{port}: {e}"),
        )),
        Err(_) => Err(OlError::new(
            ERR_EGRESS_UNREACHABLE,
            format!("could not reach {host}:{port}: connect timed out"),
        )),
    }
}

/// Open an opaque tunnel to `host:port`, through the same route
/// [`super::factory::build_proxy`] would choose for this config.
pub async fn open(cfg: &EgressConfig, host: &str, port: u16) -> Result<TunnelStream, OlError> {
    route_supported(cfg).map_err(|m| OlError::new(ERR_PROXY_SCHEME_UNSUPPORTED, m))?;

    let proxy_url = resolve_route(cfg, host, port).await?;

    let Some(route) = proxy_url else {
        return connect_direct(host, port).await;
    };
    connect_via_proxy(cfg, &route, host, port).await
}

/// Decide the proxy URL to `CONNECT` through for this one destination, or `None` for direct —
/// step 2 of `open`'s algorithm, split out so a PAC evaluation's error paths stay readable.
async fn resolve_route(
    cfg: &EgressConfig,
    host: &str,
    port: u16,
) -> Result<Option<String>, OlError> {
    if let Some(binding) = discovery::pac_binding(cfg) {
        if cfg.no_proxy.matches(host, port) {
            return Ok(None);
        }
        let target = authority_url(host, port);
        let outcome =
            tokio::task::spawn_blocking(move || discovery::pac_route_for(&target, &binding)).await;
        return match outcome {
            Ok(Ok(Some(u))) if u.scheme() == "http" => Ok(Some(u.to_string())),
            // A PAC evaluation that failed, or that legitimately answered DIRECT, degrades to
            // the same thing the forward client's own PAC closure does: go direct rather than
            // fail a request PAC never named a proxy for.
            Ok(Ok(None)) | Ok(Err(_)) | Err(_) => Ok(None),
            // Anything else — https/SOCKS from the PAC script — this transport cannot speak.
            Ok(Ok(Some(u))) => Err(OlError::new(
                ERR_PROXY_SCHEME_UNSUPPORTED,
                format!(
                    "the PAC script named a \"{}\" proxy for {host}:{port}, which the relay's opaque tunnel cannot speak",
                    u.scheme()
                ),
            )),
        };
    }
    match factory::effective_route(cfg)? {
        None => Ok(None),
        Some(route) => {
            if cfg.no_proxy.matches(host, port) {
                Ok(None)
            } else {
                Ok(Some(route))
            }
        }
    }
}

async fn connect_direct(host: &str, port: u16) -> Result<TunnelStream, OlError> {
    connect_with_timeout((host, port), host, port).await
}

async fn connect_via_proxy(
    cfg: &EgressConfig,
    route: &str,
    host: &str,
    port: u16,
) -> Result<TunnelStream, OlError> {
    let parsed = reqwest::Url::parse(route).map_err(|e| {
        OlError::new(
            ERR_PROXY_SCHEME_UNSUPPORTED,
            format!("configured proxy route \"{route}\" could not be parsed: {e}"),
        )
    })?;
    let proxy_host = parsed.host_str().ok_or_else(|| {
        OlError::new(
            ERR_PROXY_UNREACHABLE,
            format!("configured proxy route \"{route}\" names no host"),
        )
    })?;
    let proxy_port = parsed.port_or_known_default().unwrap_or(80);

    let mut stream = match tokio::time::timeout(
        TUNNEL_CONNECT_TIMEOUT,
        TcpStream::connect((proxy_host, proxy_port)),
    )
    .await
    {
        Ok(Ok(s)) => s,
        Ok(Err(e)) => {
            return Err(OlError::new(
                ERR_PROXY_UNREACHABLE,
                format!("could not reach proxy {proxy_host}:{proxy_port}: {e}"),
            ))
        }
        Err(_) => {
            return Err(OlError::new(
                ERR_PROXY_UNREACHABLE,
                format!("could not reach proxy {proxy_host}:{proxy_port}: connect timed out"),
            ))
        }
    };

    let authority = format!("{}:{port}", bracket_host(host));
    let mut request = format!("CONNECT {authority} HTTP/1.1\r\nHost: {authority}\r\n");
    if matches!(cfg.auth, ProxyAuth::Basic | ProxyAuth::Auto) {
        if let (Some(user), Some(pass)) = (cfg.username.as_deref(), cfg.proxy_password()) {
            let raw = format!("{user}:{pass}");
            let encoded = base64::engine::general_purpose::STANDARD.encode(raw.as_bytes());
            request.push_str(&format!("Proxy-Authorization: Basic {encoded}\r\n"));
        }
    }
    request.push_str("\r\n");

    stream.write_all(request.as_bytes()).await.map_err(|e| {
        OlError::new(
            ERR_PROXY_UNREACHABLE,
            format!("writing CONNECT to proxy {proxy_host}:{proxy_port} failed: {e}"),
        )
    })?;

    let head = match tokio::time::timeout(TUNNEL_HEAD_TIMEOUT, read_head(&mut stream)).await {
        Ok(Some(head)) => head,
        Ok(None) | Err(_) => {
            return Err(OlError::new(
                ERR_PROXY_UNREACHABLE,
                format!(
                    "proxy {proxy_host}:{proxy_port} did not answer the CONNECT to {host}:{port}"
                ),
            ))
        }
    };

    let status = parse_connect_status(&head);
    match status {
        Some(200) => Ok(stream),
        Some(407) => Err(OlError::new(
            ERR_PROXY_AUTH_FAILED,
            format!("proxy {proxy_host}:{proxy_port} refused authentication for {host}:{port}"),
        )),
        Some(other) => Err(OlError::new(
            ERR_PROXY_UNREACHABLE,
            format!("proxy answered CONNECT {host}:{port} with {other}"),
        )),
        None => Err(OlError::new(
            ERR_PROXY_UNREACHABLE,
            format!("proxy answered CONNECT {host}:{port} with an unparseable response"),
        )),
    }
}

/// `https://host:port/` (bracketed when `host` is an IPv6 literal) — the PAC evaluator's own
/// target URL shape (`discovery::pac_route_for`), never written to the wire.
fn authority_url(host: &str, port: u16) -> reqwest::Url {
    // Only fails on a host that cannot appear in a URL authority at all (a bare space, for
    // instance) — CONNECT already rejected such a host before this is ever called, so the
    // fallback here exists purely so this stays infallible for the caller.
    reqwest::Url::parse(&format!("https://{}:{port}/", bracket_host(host)))
        .unwrap_or_else(|_| reqwest::Url::parse("https://invalid.invalid/").unwrap())
}

#[cfg(test)]
mod tests {
    use super::super::config::ProxySource;
    use super::*;

    #[test]
    fn route_supported_accepts_direct_and_plain_http() {
        assert!(route_supported(&EgressConfig::direct()).is_ok());

        let manual = EgressConfig {
            mode: ProxyMode::Manual,
            url: Some("http://proxy.test:3128".to_string()),
            auth: ProxyAuth::Basic,
            ..EgressConfig::direct()
        };
        assert!(route_supported(&manual).is_ok());
    }

    #[test]
    fn route_supported_refuses_https_socks_negotiate_and_forbidden_direct() {
        let https_cfg = EgressConfig {
            mode: ProxyMode::Manual,
            url: Some("https://proxy.test:3128".to_string()),
            ..EgressConfig::direct()
        };
        let err = route_supported(&https_cfg).expect_err("https must be refused");
        assert!(err.contains("https"), "unexpected message: {err}");

        let socks_cfg = EgressConfig {
            mode: ProxyMode::Manual,
            url: Some("socks5://proxy.test:1080".to_string()),
            ..EgressConfig::direct()
        };
        let err = route_supported(&socks_cfg).expect_err("SOCKS must be refused");
        assert!(
            err.to_lowercase().contains("socks"),
            "unexpected message: {err}"
        );

        let negotiate_cfg = EgressConfig {
            mode: ProxyMode::Manual,
            url: Some("http://proxy.test:3128".to_string()),
            auth: ProxyAuth::Negotiate,
            ..EgressConfig::direct()
        };
        let err = route_supported(&negotiate_cfg).expect_err("Negotiate must be refused");
        assert!(err.contains("Negotiate"), "unexpected message: {err}");

        let forbidden_cfg = EgressConfig {
            mode: ProxyMode::Auto,
            url: None,
            allow_direct: false,
            ..EgressConfig::direct()
        };
        assert!(route_supported(&forbidden_cfg).is_err());

        // The PAC case (`factory::pac_with_allow_direct_false_is_refused_rather_than_leaked`
        // shape): refused on every OS, but for two different reasons.
        let pac_forbidden = EgressConfig {
            mode: ProxyMode::Auto,
            source: Some(ProxySource::Pac),
            pac_url: Some("http://wpad.test/proxy.pac".to_string()),
            allow_direct: false,
            ..EgressConfig::direct()
        };
        let err = route_supported(&pac_forbidden)
            .expect_err("a PAC-sourced route with allow_direct = false must be refused");
        if discovery::native_pac_facility().is_some() {
            assert_eq!(err, factory::PAC_DIRECT_FORBIDDEN);
        } else {
            // Linux: no binding exists, so `effective_route`'s own refusal fires instead.
            assert!(
                err.contains("allow_direct = false"),
                "unexpected message: {err}"
            );
        }

        // Control: the same PAC config with `allow_direct = true` succeeds where a facility
        // exists to bind to.
        let pac_allowed = EgressConfig {
            allow_direct: true,
            ..pac_forbidden
        };
        if discovery::native_pac_facility().is_some() {
            assert!(route_supported(&pac_allowed).is_ok());
        }
    }

    #[tokio::test]
    async fn open_direct_to_a_dead_port_is_a_typed_error() {
        let port = crate::core::egress::test_support::dead_port();
        let err = open(&EgressConfig::direct(), "127.0.0.1", port)
            .await
            .expect_err("a dead port must fail");
        assert_eq!(err.code, ERR_EGRESS_UNREACHABLE);
    }
}