openlatch-client 0.3.3

//! Egress factory.
//!
//! Every outbound byte the client sends to a non-loopback destination is meant to
//! flow through this module: one place that knows the proxy route, the TLS trust
//! source and the per-consumer timeouts: [`config`] resolves the settings, [`no_proxy`]
//! decides what bypasses, [`tls`] decides what is trusted, and [`factory`] turns all three
//! into a client for one [`Consumer`].
//!
//! Loopback is never this module's business — the hook's POST to the daemon and
//! every CLI call to an admin endpoint bypass it entirely, and the `openlatch-hook`
//! binary does not link it at all.
//!
//! [`state`] adds the runtime half — whether the route currently works — and
//! [`monitor`] is the supervised task that watches it, and that heals it.
//!
//! [`clients`] holds the swap handles the daemon's long-lived consumers read: the daemon
//! builds every client once, so a route that is only *persisted* reaches nothing, and the
//! swap is what makes a healed route apply (D-18). [`shape_memo`] keeps the
//! `proxy_configured` telemetry event rare — and carries the configuration shape only,
//! never the address (D-15).

use std::sync::Once;
use std::time::Duration;

pub mod clients;
pub mod config;
pub mod credentials;
/// The OS proxy-discovery ladder: what this host is already configured to use.
/// Distinct from [`resolve`], which decides the *auth scheme* for a route the ladder
/// (or the operator) has already chosen.
pub mod discovery;
pub mod factory;
pub mod monitor;
/// The Kerberos/SPNEGO CONNECT transport. Behind `proxy-negotiate` because the GSSAPI and
/// SSPI providers are the one place a platform security library enters the graph, so CI
/// builds it explicitly rather than by default.
#[cfg(feature = "proxy-negotiate")]
pub mod negotiate;
pub mod no_proxy;
/// `auth = "auto"`: one probe per process, before any client is built.
pub mod resolve;
pub mod shape_memo;
pub mod state;
pub mod tls;

pub use clients::{ClientHandle, EgressClients, StagedClients};
pub use config::{
    EgressConfig, EgressWarning, EnvSource, ProcessEnv, ProxyAuth, ProxyMode, ProxySource,
    ProxyToml, PROXY_TOML_KEYS,
};
pub use credentials::{
    authority_key, credential_authority, mask_userinfo, resolve_password, PasswordSource,
    ProxyCredentialFile, ProxyCredentialStore, PROXY_CREDENTIALS_FILE,
};
pub use discovery::{
    discover, pac_route_for, CandidateAttempt, CandidateOutcome, CandidateProbe, Context,
    Discovered, HealthProbe, PacAnswer, PacBinding, PacEvaluator, PacFacility, Route,
};
pub use factory::{build_blocking_client, build_client, build_client_with, Consumer, Timeouts};
pub use monitor::{run_egress_monitor, SelfHeal, HEAL_BACKOFF_INITIAL, HEAL_BACKOFF_MAX};
pub use no_proxy::{NoProxyMatcher, HARD_BYPASS};
pub use resolve::{resolve_auth, ProbeOutcome, ResolvedAuth};
pub use shape_memo::{emit_if_changed as emit_proxy_shape_if_changed, ProxyShape};
pub use state::{
    mask_text, AuthScheme, EgressReporter, EgressSnapshot, EgressState, EgressStatus, LastError,
    ProxyType, FAILURE_THRESHOLD, IDLE_PROBE_SECS,
};
pub use tls::CaSource;

/// A proxy candidate produced by discovery.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProxyCandidate {
    /// The candidate proxy URL.
    pub url: String,
    /// Where it came from.
    pub source: ProxySource,
}

/// Discovery rungs -- precedence tier 5.
///
/// I-1 defines the seam; [`resolve::OsResolver`] implements it over the per-OS ladders.
/// The trait is deliberately narrower than [`resolve::discover`]: it answers "what could
/// reach this target?" with no probing and no trace, which is what I-3's self-heal wants
/// when it is re-deciding a route it already knows is broken.
pub trait ProxyResolver: Send + Sync {
    /// Candidates for reaching `target`, best first.
    fn candidates(&self, target: &str) -> Vec<ProxyCandidate>;
}

static CRYPTO_INIT: Once = Once::new();

/// Install `ring` as the process-default rustls `CryptoProvider`.
///
/// Must run before the first `reqwest::Client` is built anywhere in the process.
/// `install_default()` succeeds at most once per process, and reqwest's
/// `rustls-no-provider` feature deliberately installs no provider of its own — a
/// client built with none installed fails at TLS setup.
///
/// Idempotent: safe to call from any entry point, any number of times.
///
/// The binaries call this at start-up *and* the constructors below call it, and both are
/// load-bearing. The constructors cover every client this crate builds, including in test
/// processes that never run `main()`. The start-up call covers what they cannot see:
/// `sentry` builds its own `reqwest::Client` inside its own crate
/// (`sentry/src/transports/reqwest.rs`), during `sentry::init`. No constructor here can
/// reach that, and neither can the `disallowed_methods` lint, which cannot inspect a
/// dependency's internals — so `crash_report::init` installs the provider itself as well.
pub fn init_crypto() {
    CRYPTO_INIT.call_once(|| {
        // An error means some other provider was installed first (feature unification
        // can pull one in via a dependency). rustls will use whichever won, so this is
        // an observation about the build graph rather than a failure: log and continue.
        if rustls::crypto::ring::default_provider()
            .install_default()
            .is_err()
        {
            tracing::warn!(
                "rustls CryptoProvider already installed; ring is not the process default"
            );
        }
    });
}

/// Start a `reqwest::ClientBuilder` with the crypto provider guaranteed installed.
///
/// Every async HTTP client in this crate is built from here. reqwest's
/// `rustls-no-provider` build panics inside `ClientBuilder::build()` when no
/// `CryptoProvider` is installed, and installing from `main()` alone is not enough:
/// unit tests -- and any other consumer of this library -- build clients without ever
/// running it. Making the install a precondition of *building a client*, rather than of
/// *starting the binary*, is what makes that class of panic unreachable.
#[allow(clippy::disallowed_methods)] // this IS the sanctioned constructor
pub fn client_builder() -> reqwest::ClientBuilder {
    init_crypto();
    reqwest::Client::builder()
}

/// Blocking counterpart of [`client_builder`], with the same precondition.
#[allow(clippy::disallowed_methods)] // this IS the sanctioned constructor
pub fn blocking_client_builder() -> reqwest::blocking::ClientBuilder {
    init_crypto();
    reqwest::blocking::Client::builder()
}

/// A default-configured client, with the crypto provider guaranteed installed.
///
/// The `reqwest::Client::new()` equivalent. It carries the same precondition as
/// [`client_builder`] -- `new()` is `builder().build().unwrap()` -- and the same panic if
/// it is not met, so it must not be called directly either.
#[allow(clippy::disallowed_methods)] // this IS the sanctioned constructor
pub fn client() -> reqwest::Client {
    init_crypto();
    reqwest::Client::new()
}

/// Blocking counterpart of [`client`].
#[allow(clippy::disallowed_methods)] // this IS the sanctioned constructor
pub fn blocking_client() -> reqwest::blocking::Client {
    init_crypto();
    reqwest::blocking::Client::new()
}

// ---------------------------------------------------------------------------
// Streaming verdict
// ---------------------------------------------------------------------------

/// How long a response may sit at "headers arrived, zero body bytes" before the
/// hop is called buffered.
///
/// **Fixed, and deliberately not configurable.** An inspection proxy that
/// reassembles a stream holds it for the whole generation — tens of seconds —
/// while a working SSE hop emits its first byte in well under a second. There is
/// no operator judgement in that gap worth a knob, and a knob would only ever be
/// turned to make a real buffering problem stop being reported.
///
/// [`stream_probe`] is the only entry point, and it hardcodes this value, so no
/// shipped path can see any other. The tests reach a crate-private variant that
/// takes the threshold instead — which is why the constant itself needs no
/// `cfg(test)` alternate.
pub const STREAM_BUFFERED_AFTER: Duration = Duration::from_secs(10);

/// What a streaming request through the chain turned out to be.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StreamVerdict {
    /// Body bytes arrived promptly after the headers. The hop streams.
    Streaming,
    /// Headers arrived and the body stayed empty past [`STREAM_BUFFERED_AFTER`].
    /// Something in the path is reassembling the response (`OL-1228`).
    Buffered,
    /// No valid streaming target was available, so nothing was attempted.
    ///
    /// **The production verdict.** The client holds no provider credential, and a
    /// credential-less POST to `api.anthropic.com` is answered `401` without ever
    /// streaming — which classifies the proxy, not the stream. The streaming leg
    /// runs for real against the e2e fixture.
    Skipped,
    /// The request did not get far enough to say. A refused connection, a non-2xx
    /// answer, or a stream that closed with no body at all.
    Unclassified,
}

impl StreamVerdict {
    /// The wire string for `proxy test --json`.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Streaming => "streaming",
            Self::Buffered => "buffered",
            Self::Skipped => "skipped",
            Self::Unclassified => "unclassified",
        }
    }
}

/// The streaming leg's result, shaped for the `proxy test` contract.
#[derive(Debug, Clone)]
pub struct StreamProbe {
    /// See [`StreamVerdict`].
    pub verdict: StreamVerdict,
    /// The `OL-122x` for a verdict that carries one. Only `Buffered` does.
    pub code: Option<&'static str>,
    /// Milliseconds from request start to the first body byte, when there was one.
    pub first_byte_ms: Option<u64>,
    /// One line of context, userinfo masked. Never a bare socket error.
    pub detail: Option<String>,
}

impl StreamProbe {
    fn verdict_only(verdict: StreamVerdict) -> Self {
        Self {
            verdict,
            code: None,
            first_byte_ms: None,
            detail: None,
        }
    }

    fn unclassified(detail: impl Into<String>) -> Self {
        Self {
            verdict: StreamVerdict::Unclassified,
            code: None,
            first_byte_ms: None,
            detail: Some(detail.into()),
        }
    }
}

/// Classify one streaming request through the configured chain.
///
/// `target` is `None` in production — see [`StreamVerdict::Skipped`] — and a real
/// URL under the e2e fixture and the tests below.
///
/// **I-2 gated:** `openlatch proxy test` is this function's only production
/// caller, and `src/cli/commands/proxy.rs` does not exist on this branch — I-2
/// creates it. When it lands, the `stream` leg of `proxy test` calls this in
/// place of its `unclassified` shim and renders
/// `{ verdict, code? }` from the returned [`StreamProbe`] per the frozen
/// `proxy test --json` table. Nothing else has to change here.
///
/// `client` must not carry a whole-request timeout shorter than
/// [`STREAM_BUFFERED_AFTER`], or the buffering case would surface as a client
/// timeout instead of as `OL-1228`. [`Consumer::Boundary`] is the preset built
/// for exactly this shape: bounded connect, no total deadline.
pub async fn stream_probe(client: &reqwest::Client, target: Option<&str>) -> StreamProbe {
    stream_probe_within(client, target, STREAM_BUFFERED_AFTER).await
}

/// [`stream_probe`] with an explicit threshold.
///
/// Crate-private on purpose: the fixed ten seconds stays the only value any
/// shipped path can reach, and the tests drive the classifier in milliseconds
/// rather than adding ten seconds to the suite.
pub(crate) async fn stream_probe_within(
    client: &reqwest::Client,
    target: Option<&str>,
    buffered_after: Duration,
) -> StreamProbe {
    let Some(target) = target else {
        return StreamProbe::verdict_only(StreamVerdict::Skipped);
    };

    let started = std::time::Instant::now();
    let mut response = match client
        .get(target)
        .header(reqwest::header::ACCEPT, "text/event-stream")
        .send()
        .await
    {
        Ok(r) => r,
        Err(e) => return StreamProbe::unclassified(state::mask_text(&e.to_string())),
    };

    let status = response.status();
    if !status.is_success() {
        return StreamProbe::unclassified(format!("upstream answered {status}"));
    }

    // Headers are in. From here the only question is whether a body byte shows
    // up before the threshold — that, and nothing about content, is what
    // separates a streaming hop from a reassembling one.
    let deadline = tokio::time::Instant::now() + buffered_after;
    loop {
        match tokio::time::timeout_at(deadline, response.chunk()).await {
            // A body byte inside the window: the hop streams.
            Ok(Ok(Some(chunk))) if !chunk.is_empty() => {
                return StreamProbe {
                    verdict: StreamVerdict::Streaming,
                    code: None,
                    first_byte_ms: Some(
                        started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64
                    ),
                    detail: None,
                };
            }
            // An empty frame is not a byte. Keep waiting on the same deadline.
            Ok(Ok(Some(_))) => continue,
            // The body ended without ever producing one. That is a strange
            // upstream, not a buffering proxy — do not blame the network for it.
            Ok(Ok(None)) => {
                return StreamProbe::unclassified("upstream closed the body with no bytes")
            }
            Ok(Err(e)) => return StreamProbe::unclassified(state::mask_text(&e.to_string())),
            Err(_elapsed) => {
                return StreamProbe {
                    verdict: StreamVerdict::Buffered,
                    code: Some(crate::core::error::ERR_STREAM_BUFFERED),
                    first_byte_ms: None,
                    detail: Some(format!(
                        "headers arrived but no body byte within {}s",
                        buffered_after.as_secs_f32()
                    )),
                };
            }
        }
    }
}

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

    #[test]
    fn masking_removes_the_password_and_keeps_everything_else() {
        assert_eq!(
            mask_userinfo("http://alice:s3cr3t@proxy.corp:8080"),
            "http://alice:*****@proxy.corp:8080"
        );
        // A `@` inside the password must not split the authority early.
        assert_eq!(
            mask_userinfo("http://alice:p@ss@proxy.corp:8080"),
            "http://alice:*****@proxy.corp:8080"
        );
        // No password: inventing one would look like a credential that is not there.
        assert_eq!(
            mask_userinfo("http://alice@proxy.corp:8080"),
            "http://alice@proxy.corp:8080"
        );
        // The common case, untouched.
        assert_eq!(
            mask_userinfo("http://proxy.corp:8080"),
            "http://proxy.corp:8080"
        );
        assert_eq!(mask_userinfo(""), "");
        assert_eq!(mask_userinfo("pac:wpad"), "pac:wpad");
        assert_eq!(
            mask_userinfo("http://alice:s3cr3t@proxy.corp:8080/path"),
            "http://alice:*****@proxy.corp:8080/path"
        );
    }

    /// The provider is what makes a TLS handshake possible at all. Reaching *certificate
    /// verification* against a locally-served self-signed leaf therefore proves the whole
    /// chain is wired: `ring` is installed (no provider means the client fails before any
    /// bytes move) and reqwest's `rustls-no-provider` build completed a real handshake up
    /// to the trust decision. It does NOT prove *which* trust store rejected the cert —
    /// any verifier rejects a self-signed leaf — so this test claims only the crypto path.
    #[tokio::test]
    async fn a_real_handshake_reaches_certificate_verification() {
        use std::sync::Arc;
        use tokio::net::TcpListener;

        init_crypto();

        let issued = rcgen::generate_simple_self_signed(vec!["localhost".to_string()])
            .expect("generate self-signed cert");
        let server_config = tokio_rustls::rustls::ServerConfig::builder()
            .with_no_client_auth()
            .with_single_cert(
                vec![issued.cert.der().clone()],
                rustls::pki_types::PrivateKeyDer::Pkcs8(issued.signing_key.serialize_der().into()),
            )
            .expect("server config");

        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let port = listener.local_addr().expect("local_addr").port();
        let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(server_config));
        // Joined below: a connection that never arrives is exactly the failure this test
        // exists to rule out, so the server side has to be able to say whether one did.
        let server = tokio::spawn(async move {
            let Ok((stream, _)) = listener.accept().await else {
                return false;
            };
            // The client is expected to reject the certificate, so the handshake fails on
            // this side too. Reaching this call at all is the proof: TCP connected and the
            // server got far enough to present a certificate.
            let _ = acceptor.accept(stream).await;
            true
        });

        let client = client_builder()
            .timeout(std::time::Duration::from_secs(5))
            .build()
            .expect("build client");
        // 127.0.0.1 literally, never "localhost": on Windows that resolves ::1 first,
        // nothing is listening there, and the resulting connection error would satisfy a
        // loose assertion without a single TLS byte having moved.
        let err = client
            .get(format!("https://127.0.0.1:{port}/"))
            .send()
            .await
            .expect_err("a self-signed leaf must not verify");

        assert!(
            server.await.expect("server task"),
            "no connection reached the TLS server -- this test proved nothing"
        );
        assert!(
            err.is_connect() || err.is_request(),
            "expected a certificate failure, got: {err}"
        );
    }

    // --- streaming verdict ---------------------------------------------------

    /// The client the streaming leg needs: bounded connect, no total deadline.
    /// A client with a total timeout under the threshold would turn the
    /// buffering case into a client-side timeout and hide `OL-1228`.
    fn stream_client() -> reqwest::Client {
        build_client(Consumer::Boundary, &EgressConfig::direct()).expect("stream client")
    }

    /// A mock that sends response headers and then holds the body open, saying
    /// nothing — the shape an inspecting proxy produces while it reassembles.
    async fn spawn_buffering_upstream(hold: Duration) -> u16 {
        use tokio::io::AsyncWriteExt;
        let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
            .await
            .expect("bind");
        let port = listener.local_addr().expect("addr").port();
        tokio::spawn(async move {
            if let Ok((mut stream, _)) = listener.accept().await {
                // Read whatever the client sends; we never answer its content.
                let mut scratch = [0u8; 1024];
                let _ = tokio::io::AsyncReadExt::read(&mut stream, &mut scratch).await;
                let head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n\
                            Transfer-Encoding: chunked\r\n\r\n";
                let _ = stream.write_all(head.as_bytes()).await;
                let _ = stream.flush().await;
                tokio::time::sleep(hold).await;
                let _ = stream.write_all(b"0\r\n\r\n").await;
            }
        });
        port
    }

    #[test]
    fn the_buffering_threshold_is_ten_seconds_and_takes_no_knob() {
        assert_eq!(STREAM_BUFFERED_AFTER, Duration::from_secs(10));
    }

    #[tokio::test]
    async fn no_target_is_skipped_without_touching_the_network() {
        let probe = stream_probe(&stream_client(), None).await;
        assert_eq!(probe.verdict, StreamVerdict::Skipped);
        assert_eq!(probe.code, None);
        assert_eq!(probe.verdict.as_str(), "skipped");
    }

    #[tokio::test]
    async fn a_trickling_upstream_is_streaming() {
        let upstream = crate::boundary::mock::spawn_trickle_sse(3, Duration::from_millis(20)).await;
        // A generous threshold on purpose: the classifier returns the instant the
        // first byte lands, so a large window cannot slow the test down — it only
        // removes any chance of a loaded machine being mistaken for a proxy.
        let probe = stream_probe_within(
            &stream_client(),
            Some(&format!("http://127.0.0.1:{}/v1/messages", upstream.port)),
            Duration::from_secs(10),
        )
        .await;
        assert_eq!(
            probe.verdict,
            StreamVerdict::Streaming,
            "first byte arrived well inside the window: {:?}",
            probe.detail
        );
        assert!(probe.first_byte_ms.is_some());
        assert_eq!(probe.code, None);
    }

    #[tokio::test]
    async fn an_upstream_that_holds_the_body_is_buffered() {
        let threshold = Duration::from_millis(200);
        let port = spawn_buffering_upstream(Duration::from_secs(5)).await;
        let probe = stream_probe_within(
            &stream_client(),
            Some(&format!("http://127.0.0.1:{port}/v1/messages")),
            threshold,
        )
        .await;
        assert_eq!(probe.verdict, StreamVerdict::Buffered);
        assert_eq!(probe.code, Some(crate::core::error::ERR_STREAM_BUFFERED));
        assert_eq!(probe.verdict.as_str(), "buffered");
    }

    #[tokio::test]
    async fn an_unreachable_target_is_unclassified_not_buffered() {
        // Nothing is listening. This must never be reported as a buffering
        // proxy — the remedy for OL-1228 is an inspection exemption, which
        // would be exactly the wrong thing to ask for here.
        let probe = stream_probe(&stream_client(), Some("http://127.0.0.1:1/v1/messages")).await;
        assert_eq!(probe.verdict, StreamVerdict::Unclassified);
        assert_eq!(probe.code, None);
        assert!(probe.detail.is_some(), "a verdict must say why");
    }

    #[test]
    fn init_crypto_is_idempotent() {
        init_crypto();
        init_crypto();
        init_crypto();
        // A provider must be resolvable afterwards; without one, rustls returns None
        // and every TLS client build in the process would fail.
        assert!(
            rustls::crypto::CryptoProvider::get_default().is_some(),
            "no process-default CryptoProvider after init_crypto()"
        );
    }
}