openlatch-client 0.3.3

//! Per-consumer client construction.
//!
//! One function builds every outbound client in the product, so the proxy route, the trust
//! source and the timeouts are decided in exactly one place. The timeout presets reproduce
//! today's values verbatim: this plan changes the route traffic takes, never how long a
//! consumer is willing to wait for it.

use std::time::Duration;

use crate::core::error::{OlError, ERR_DIRECT_FORBIDDEN, ERR_PROXY_SCHEME_UNSUPPORTED};

use super::config::{EgressConfig, ProxyAuth, ProxyMode};
use super::{blocking_client_builder, client_builder, tls};

/// Which call site a client is being built for.
///
/// One variant per row of the egress inventory, so no consumer can silently inherit
/// another's timeouts.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Consumer {
    /// The cloud event worker.
    CloudWorker,
    /// The policy-bundle poller.
    PolicyPoller,
    /// The pending-alerts long poll.
    Alerts,
    /// Auth: key validation and revocation.
    Auth,
    /// `status`'s cloud reachability probe. Blocking.
    StatusProbe,
    /// The update manifest check.
    UpdateCheck,
    /// The update tarball download.
    UpdateDownload,
    /// PostHog telemetry.
    Telemetry,
    /// The model boundary's upstream forwarder.
    Boundary,
}

struct Preset {
    connect: Option<Duration>,
    total: Option<Duration>,
    pool_idle: Option<usize>,
}

/// Deadlines a consumer supplies for itself, in place of its preset's.
///
/// Four presets leave `connect` and `total` unset on purpose: the cloud worker takes both
/// from `CloudConfig`, the two pollers take theirs from the deployment's cloud timeout,
/// and the tarball download from the size of what it is fetching. This is how they hand
/// the value back without any of them reaching for a raw `ClientBuilder`.
#[derive(Debug, Clone, Copy, Default)]
pub struct Timeouts {
    /// TCP connect deadline. `None` keeps the preset's.
    pub connect: Option<Duration>,
    /// Whole-request deadline. `None` keeps the preset's.
    pub total: Option<Duration>,
}

impl Timeouts {
    /// A whole-request deadline only, leaving connect to the preset.
    pub fn total(d: Duration) -> Self {
        Self {
            connect: None,
            total: Some(d),
        }
    }
}

impl Consumer {
    fn preset(self) -> Preset {
        match self {
            // The boundary is the one consumer with no total timeout and no read timeout,
            // and that is deliberate. It bounds response *headers* only, at 60s, from the
            // caller side. A per-read timeout would kill a legitimate SSE stream that is
            // merely quiet while the model is thinking -- exactly the regression this must
            // not introduce.
            Self::Boundary => Preset {
                connect: Some(Duration::from_secs(10)),
                total: None,
                pool_idle: Some(8),
            },
            // The cloud worker's timeouts are per-deployment and come from CloudConfig, so
            // the caller applies them; only the pool shape is fixed here.
            Self::CloudWorker => Preset {
                connect: None,
                total: None,
                pool_idle: Some(4),
            },
            Self::Auth => Preset {
                connect: None,
                total: Some(Duration::from_secs(10)),
                pool_idle: None,
            },
            Self::StatusProbe => Preset {
                connect: None,
                total: Some(Duration::from_secs(3)),
                pool_idle: None,
            },
            Self::UpdateCheck => Preset {
                connect: None,
                total: Some(Duration::from_secs(5)),
                pool_idle: None,
            },
            // Poller, alerts and download all take their deadline from the caller, which
            // knows the deployment's configured value or the size of what it is fetching.
            Self::PolicyPoller | Self::Alerts | Self::UpdateDownload | Self::Telemetry => Preset {
                connect: None,
                total: None,
                pool_idle: None,
            },
        }
    }
}

/// Decide whether this configuration routes through a proxy at all.
///
/// `Ok(None)` means direct. The error case is the one where a host would otherwise go
/// direct silently, which some security teams read as an exfiltration signal.
pub(super) fn effective_route(cfg: &EgressConfig) -> Result<Option<String>, OlError> {
    if cfg.mode == ProxyMode::Direct {
        return Ok(None);
    }
    match cfg.url.as_deref() {
        Some(u) => Ok(Some(u.to_string())),
        None if cfg.allow_direct => Ok(None),
        None => Err(OlError::new(
            ERR_DIRECT_FORBIDDEN,
            "no proxy is configured, and [proxy] allow_direct = false forbids going direct",
        )
        .with_suggestion(
            "Set [proxy] url, or allow a direct connection with allow_direct = true.",
        )),
    }
}

/// Embed Basic credentials in the proxy URL when we have them.
///
/// reqwest treats a 407 as terminal, so Basic cannot be negotiated reactively: it is
/// presented on the first request or not at all.
fn proxy_url_with_auth(cfg: &EgressConfig, route: &str) -> String {
    // Only Basic rides the URL. `Negotiate` reaches this function once `resolve_auth` has
    // concretized `auto` -- and embedding userinfo there would put a Basic
    // `Proxy-Authorization` on the wire to a proxy that asked for Kerberos, which is both
    // wrong and a credential disclosed to a peer that never requested it.
    //
    // `Auto` still embeds, because an unresolved config is one that never went through
    // `resolve_auth` (a unit test, a library consumer, a client built before start-up
    // finished) and preemptive Basic is what it did before this transport existed.
    if !matches!(cfg.auth, ProxyAuth::Basic | ProxyAuth::Auto) {
        return route.to_string();
    }
    // `proxy_password()` is the whole credential ladder -- env, keychain, file -- collapsed
    // to one value by `resolve_auth`. Reading `env_password` directly here would silently
    // ignore a password that lives in the OS keychain, which is where most of them live.
    let (Some(user), Some(pass)) = (cfg.username.as_deref(), cfg.proxy_password()) else {
        return route.to_string();
    };
    match route.split_once("://") {
        Some((scheme, rest)) => format!(
            "{scheme}://{}:{}@{rest}",
            percent_encode_userinfo(user),
            percent_encode_userinfo(pass)
        ),
        None => route.to_string(),
    }
}

/// Percent-encode the characters that would otherwise re-split the authority.
fn percent_encode_userinfo(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for b in s.bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
                out.push(b as char);
            }
            _ => out.push_str(&format!("%{b:02X}")),
        }
    }
    out
}

/// Build the `reqwest::Proxy` for a route, or `None` for direct.
///
/// The bypass decision lives inside the closure because that is the only place it is ever
/// consulted: reqwest's own `NoProxy`, attached to a custom proxy, is a dead field, and its
/// grammar is curl's rather than the Go grammar this product specifies.
fn build_proxy(cfg: &EgressConfig) -> Result<Option<reqwest::Proxy>, OlError> {
    // A PAC-sourced configuration carries no `url` by contract, so it must be handled
    // before `effective_route` reads that emptiness as "go direct". PAC answers per
    // destination, so the route is decided inside the closure, once per request, against
    // the OS evaluator. An evaluation that fails leaves this request direct rather than
    // failing it: a PAC outage is a network state, and network states never fail closed
    // here (D-9).
    if let Some(binding) = super::discovery::pac_binding(cfg) {
        // `allow_direct = false` and PAC cannot both be honoured, and the reason is
        // structural rather than a missing feature: a custom-proxy closure can answer
        // "this proxy" or "direct", and has no way to *fail* a request. A PAC whose
        // evaluation failed — or that legitimately answers DIRECT — would therefore leave
        // the request unproxied, which is the one outcome this flag exists to forbid.
        // Naming the contradiction beats quietly leaking traffic past it.
        if !cfg.allow_direct {
            return Err(OlError::new(
                ERR_DIRECT_FORBIDDEN,
                "a PAC-sourced proxy cannot guarantee allow_direct = false: a PAC may                  answer DIRECT, and a failed evaluation has no route to fall back to",
            )
            .with_suggestion(
                "Set [proxy] url to the proxy the PAC returns for this host, or set                  allow_direct = true.",
            ));
        }
        let matcher = cfg.no_proxy.clone();
        return Ok(Some(reqwest::Proxy::custom(move |dst| {
            let host = dst.host_str()?;
            let port = dst.port_or_known_default()?;
            if matcher.matches(host, port) {
                return None;
            }
            super::discovery::pac_route_for(dst, &binding)
                .ok()
                .flatten()
        })));
    }
    let Some(route) = effective_route(cfg)? else {
        return Ok(None);
    };
    let target = proxy_url_with_auth(cfg, &route);
    let matcher = cfg.no_proxy.clone();
    Ok(Some(reqwest::Proxy::custom(move |dst| {
        let host = dst.host_str()?;
        let port = dst.port_or_known_default()?;
        if matcher.matches(host, port) {
            return None;
        }
        target.parse::<reqwest::Url>().ok()
    })))
}

/// Reject a scheme this build cannot speak, rather than discovering it at request time.
fn check_scheme(cfg: &EgressConfig) -> Result<(), OlError> {
    if cfg.auth == ProxyAuth::Negotiate && !cfg!(feature = "proxy-negotiate") {
        return Err(OlError::new(
            ERR_PROXY_SCHEME_UNSUPPORTED,
            "[proxy] auth = negotiate needs a build with the proxy-negotiate feature",
        )
        .with_suggestion(
            "Release binaries ship the feature; a local build needs \
             --features proxy-negotiate.",
        ));
    }
    Ok(())
}

fn build_failed(consumer: Consumer, e: reqwest::Error) -> OlError {
    OlError::new(
        ERR_PROXY_SCHEME_UNSUPPORTED,
        format!("could not build the {consumer:?} http client: {e}"),
    )
    .with_suggestion("Check [proxy] url and ca_bundle in config.toml.")
}

/// Build the async client for one consumer.
pub fn build_client(consumer: Consumer, cfg: &EgressConfig) -> Result<reqwest::Client, OlError> {
    build_client_with(consumer, cfg, Timeouts::default())
}

/// [`build_client`], with the caller's own deadlines overriding the preset's.
pub fn build_client_with(
    consumer: Consumer,
    cfg: &EgressConfig,
    timeouts: Timeouts,
) -> Result<reqwest::Client, OlError> {
    check_scheme(cfg)?;
    let p = consumer.preset();
    let mut b = client_builder();
    if let Some(d) = timeouts.connect.or(p.connect) {
        b = b.connect_timeout(d);
    }
    if let Some(d) = timeouts.total.or(p.total) {
        b = b.timeout(d);
    }
    if let Some(n) = p.pool_idle {
        b = b.pool_max_idle_per_host(n);
    }
    if cfg.http1_only {
        b = b.http1_only();
    }
    b = tls::apply(b, cfg)?;
    // `no_proxy()` first, always: it is what stops reqwest reading the ambient environment
    // on its own. Being explicit in both directions means the route is ours, never inherited.
    b = b.no_proxy();
    if let Some(proxy) = build_proxy(cfg)? {
        b = b.proxy(proxy);
    }
    b.build().map_err(|e| build_failed(consumer, e))
}

/// Build the blocking client for one consumer.
pub fn build_blocking_client(
    consumer: Consumer,
    cfg: &EgressConfig,
) -> Result<reqwest::blocking::Client, OlError> {
    check_scheme(cfg)?;
    let p = consumer.preset();
    let mut b = blocking_client_builder();
    if let Some(d) = p.connect {
        b = b.connect_timeout(d);
    }
    if let Some(d) = p.total {
        b = b.timeout(d);
    }
    if let Some(n) = p.pool_idle {
        b = b.pool_max_idle_per_host(n);
    }
    if cfg.http1_only {
        b = b.http1_only();
    }
    b = tls::apply_blocking(b, cfg)?;
    b = b.no_proxy();
    if let Some(proxy) = build_proxy(cfg)? {
        b = b.proxy(proxy);
    }
    b.build().map_err(|e| build_failed(consumer, e))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::egress::config::ProxyToml;
    use std::collections::HashMap;

    struct MapEnv(HashMap<String, String>);
    impl super::super::config::EnvSource for MapEnv {
        fn var(&self, key: &str) -> Option<String> {
            self.0.get(key).cloned()
        }
    }
    fn empty_env() -> MapEnv {
        MapEnv(HashMap::new())
    }

    fn cfg_with(url: Option<&str>) -> EgressConfig {
        let toml = ProxyToml {
            url: url.map(str::to_string),
            ..Default::default()
        };
        EgressConfig::resolve(Some(&toml), &empty_env(), 7443, 7444).expect("resolve")
    }

    #[test]
    fn every_consumer_builds() {
        let cfg = cfg_with(Some("http://proxy.test:8080"));
        for c in [
            Consumer::CloudWorker,
            Consumer::PolicyPoller,
            Consumer::Alerts,
            Consumer::Auth,
            Consumer::UpdateCheck,
            Consumer::UpdateDownload,
            Consumer::Telemetry,
            Consumer::Boundary,
        ] {
            build_client(c, &cfg).unwrap_or_else(|e| panic!("{c:?} failed to build: {e:?}"));
        }
        build_blocking_client(Consumer::StatusProbe, &cfg).expect("status probe builds");
    }

    #[test]
    fn direct_config_builds_every_consumer_too() {
        let cfg = EgressConfig::direct();
        for c in [Consumer::Boundary, Consumer::CloudWorker, Consumer::Auth] {
            build_client(c, &cfg).unwrap_or_else(|e| panic!("{c:?} failed to build: {e:?}"));
        }
    }

    #[test]
    fn the_boundary_preset_keeps_its_header_only_posture() {
        // The regression this guards: adding a total or per-read timeout here would cut
        // off a long, legitimately quiet SSE stream.
        let p = Consumer::Boundary.preset();
        assert_eq!(p.connect, Some(Duration::from_secs(10)));
        assert_eq!(p.total, None, "the boundary must have NO total timeout");
        assert_eq!(p.pool_idle, Some(8));
    }

    #[test]
    fn presets_match_the_inventory() {
        assert_eq!(Consumer::Auth.preset().total, Some(Duration::from_secs(10)));
        assert_eq!(
            Consumer::StatusProbe.preset().total,
            Some(Duration::from_secs(3))
        );
        assert_eq!(
            Consumer::UpdateCheck.preset().total,
            Some(Duration::from_secs(5))
        );
        assert_eq!(Consumer::CloudWorker.preset().pool_idle, Some(4));
    }

    #[test]
    fn caller_deadlines_override_the_preset_and_nothing_else() {
        // The four presets that set no deadline would otherwise migrate their call sites
        // into a client with no timeout at all -- a long poll against a dead peer would
        // then hang forever instead of failing at the deployment's configured budget.
        let cfg = cfg_with(Some("http://proxy.test:8080"));
        build_client_with(
            Consumer::PolicyPoller,
            &cfg,
            Timeouts::total(Duration::from_secs(30)),
        )
        .expect("poller builds with a caller deadline");
        assert_eq!(Timeouts::total(Duration::from_secs(30)).connect, None);
        assert_eq!(Timeouts::default().total, None);
    }

    #[test]
    fn direct_mode_routes_nowhere() {
        let cfg = EgressConfig::direct();
        assert_eq!(effective_route(&cfg).expect("route"), None);
    }

    #[test]
    fn allow_direct_false_without_a_proxy_is_refused() {
        let toml = ProxyToml {
            allow_direct: Some(false),
            ..Default::default()
        };
        let cfg = EgressConfig::resolve(Some(&toml), &empty_env(), 7443, 7444).expect("resolve");
        let err = effective_route(&cfg).expect_err("must refuse");
        assert_eq!(err.code, ERR_DIRECT_FORBIDDEN);
    }

    #[test]
    fn basic_credentials_are_embedded_and_encoded() {
        let mut cfg = cfg_with(Some("http://proxy.test:8080"));
        cfg.username = Some("dom\\alice".into());
        cfg.env_password = Some("p@ss word".into());
        let url = proxy_url_with_auth(&cfg, "http://proxy.test:8080");
        assert_eq!(url, "http://dom%5Calice:p%40ss%20word@proxy.test:8080");
    }

    /// The Negotiate transport authenticates in its own CONNECT exchange. A reqwest client
    /// built from a Negotiate-resolved config must not additionally embed Basic userinfo:
    /// that would disclose the password to a proxy that asked for Kerberos.
    #[test]
    fn negotiate_never_embeds_basic_credentials() {
        let mut cfg = cfg_with(Some("http://proxy.test:8080"));
        cfg.username = Some("alice".into());
        cfg.env_password = Some("secret".into());
        cfg.auth = ProxyAuth::Negotiate;
        assert_eq!(
            proxy_url_with_auth(&cfg, "http://proxy.test:8080"),
            "http://proxy.test:8080"
        );
    }

    #[test]
    fn auth_none_never_embeds_credentials() {
        let mut cfg = cfg_with(Some("http://proxy.test:8080"));
        cfg.username = Some("alice".into());
        cfg.env_password = Some("secret".into());
        cfg.auth = ProxyAuth::None;
        assert_eq!(
            proxy_url_with_auth(&cfg, "http://proxy.test:8080"),
            "http://proxy.test:8080"
        );
    }

    #[test]
    fn a_pac_sourced_config_installs_a_proxy_closure_instead_of_going_direct() {
        // The regression this pins: a PAC-sourced config carries no `url` by contract, and
        // before the resolver landed `effective_route` read that emptiness as "go direct" —
        // routing an enterprise's traffic around the very proxy its PAC names.
        let toml = ProxyToml {
            source: Some("pac".to_string()),
            pac_url: Some("http://wpad.corp/proxy.pac".to_string()),
            ..Default::default()
        };
        let cfg = EgressConfig::resolve(Some(&toml), &empty_env(), 7443, 7444).expect("resolve");
        assert!(cfg.url.is_none(), "a PAC source must persist no url");
        let proxy = build_proxy(&cfg).expect("build");
        if crate::core::egress::discovery::native_pac_facility().is_some() {
            assert!(
                proxy.is_some(),
                "a PAC-capable host must get a per-destination closure"
            );
        } else {
            // Linux: D-20, there is no evaluator to bind to, so the config falls through
            // to whatever static route it declares — here, none.
            assert!(proxy.is_none());
        }
        // Whatever the platform, building a real client must still succeed: a PAC that
        // cannot be evaluated is a network state, and network states never fail startup.
        build_client(Consumer::CloudWorker, &cfg).expect("a PAC config still builds a client");
    }

    #[test]
    fn pac_with_allow_direct_false_is_refused_rather_than_leaked() {
        // The leak this closes: the closure would answer `None` on a failed evaluation,
        // and `None` means direct — on a host whose policy is that nothing leaves
        // unproxied.
        let toml = ProxyToml {
            source: Some("pac".to_string()),
            pac_url: Some("http://wpad.corp/proxy.pac".to_string()),
            allow_direct: Some(false),
            ..Default::default()
        };
        let cfg = EgressConfig::resolve(Some(&toml), &empty_env(), 7443, 7444).expect("resolve");
        let err = build_proxy(&cfg);
        if crate::core::egress::discovery::native_pac_facility().is_some() {
            let err = err.expect_err("the combination must be refused, not silently allowed");
            assert_eq!(err.code, ERR_DIRECT_FORBIDDEN);
            assert!(err.suggestion.is_some(), "a refusal owes a remedy");
        } else {
            // No PAC facility, so no binding: the existing `effective_route` refusal is
            // what fires, with the same code.
            assert_eq!(
                err.expect_err("no proxy and no direct is still refused")
                    .code,
                ERR_DIRECT_FORBIDDEN
            );
        }
    }

    #[test]
    fn a_manual_source_is_never_treated_as_a_pac_binding() {
        let mut cfg = cfg_with(Some("http://proxy.test:8080"));
        cfg.source = Some(crate::core::egress::ProxySource::Manual);
        cfg.pac_url = Some("http://wpad.corp/proxy.pac".to_string());
        assert!(
            crate::core::egress::discovery::pac_binding(&cfg).is_none(),
            "a stray pac_url must not hijack a static route"
        );
    }

    #[test]
    fn negotiate_without_the_feature_is_named_not_silently_downgraded() {
        let mut cfg = cfg_with(Some("http://proxy.test:8080"));
        cfg.auth = ProxyAuth::Negotiate;
        if cfg!(feature = "proxy-negotiate") {
            assert!(check_scheme(&cfg).is_ok());
        } else {
            let err = check_scheme(&cfg).expect_err("must refuse");
            assert_eq!(err.code, ERR_PROXY_SCHEME_UNSUPPORTED);
        }
    }
}