openlatch-client 0.3.3

//! The Negotiate (Kerberos/SPNEGO) proxy transport.
//!
//! reqwest cannot host this. Its connector seam is sealed — `Proxy::custom` returns a URL
//! and `connector_layer` cannot see the CONNECT exchange — and a 407 is terminal there, so
//! a multi-leg SPNEGO handshake is structurally impossible inside it. The Negotiate path is
//! therefore its own connector, driving its own CONNECT tunnel, behind the `proxy-negotiate`
//! feature.
//!
//! What [`NegotiateConnector::connect`] does, in order — the order is the design:
//!
//! | # | Step | Why it is where it is |
//! | - | ---- | --------------------- |
//! | 0 | **`no_proxy` first** | reqwest's bypass lives inside a closure that does not exist on this transport. Without this step, loopback and every operator `no_proxy` entry would start being proxied the moment Negotiate is selected — a silently different bypass policy per transport. |
//! | 1 | TCP connect to the proxy | |
//! | 2 | TLS **to the proxy** for an `https://` proxy, ALPN excluded | The token is only ever written inside this TLS. ALPN is excluded because an `h2` proxy hop would break the HTTP/1.1 CONNECT writer below it. |
//! | 3 | Mint a **fresh** SPNEGO token | Squid's and MIT's replay caches reject a replayed token, so a pooled tunnel must already be authenticated. Minting inside `connect` is what guarantees the pool never holds an unauthenticated one. |
//! | 4 | `CONNECT` + `Proxy-Authorization: Negotiate <b64>` | |
//! | 5 | Read the response, and **continue the exchange if it says to** | A 407 carrying `Proxy-Authenticate: Negotiate <token>` while the provider says continue is a *continuation on this same TCP connection*, not a rejection. A 200 may carry a final token for mutual auth, which the provider must consume before the tunnel is handed up. |
//! | 6 | TLS **to the target**, only for an `https` destination | A plain-HTTP `api_url` is legal in development, and so are the plain mock upstreams. ALPN is restored here. |
//!
//! **D-5, precisely.** "A 407 is terminal" governs a *concluded* rejection: a bare 407, a
//! challenge this build cannot use, a provider that says it is done, or the leg budget
//! exhausted. Each of those ends the connection with no retry. A 407 that carries a
//! continuation token is none of those things — it is the middle of one handshake.
//!
//! **NTLM is refused, never fallen back to** (D-2). On Windows the provider asserts the
//! negotiated package is Kerberos before it releases the first token, so a workgroup machine
//! never puts one on the wire. On Unix the SPNEGO mech OID with krb5 makes NTLM
//! structurally unavailable.

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;

use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf};
use tokio::net::TcpStream;

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

use super::config::EgressConfig;
use super::credentials::mask_userinfo;
use super::no_proxy::NoProxyMatcher;

mod tls;

#[cfg(unix)]
pub mod gssapi;
#[cfg(windows)]
pub mod sspi;

pub use tls::TlsSetup;

/// How many CONNECT legs one tunnel may take before we call it a loop.
///
/// A real SPNEGO exchange concludes in one or two. Four leaves room for a proxy that
/// inserts a round trip of its own, and stops a misbehaving one from spinning forever
/// against a provider that keeps saying "continue".
pub const MAX_LEGS: usize = 4;

/// Ceiling on the proxy's response head. A proxy that will not finish a header block in
/// 8 KiB is either broken or trying something.
const MAX_RESPONSE_HEAD: usize = 8 * 1024;

// ---------------------------------------------------------------------------
// Provider seam
// ---------------------------------------------------------------------------

/// One leg of a SPNEGO exchange.
#[derive(Debug)]
pub enum StepResult {
    /// Send this token, then expect another challenge.
    Continue(Vec<u8>),
    /// The exchange is concluded on our side. `Some(token)` still has to be sent.
    Done(Option<Vec<u8>>),
    /// Refused. Nothing is written to the wire.
    Failed(NegotiateError),
}

/// A SPNEGO security context, driven one leg at a time.
///
/// `peer` is the token from the previous challenge, `None` on the first leg. Modelling this
/// as a state machine rather than a one-shot `token()` call is what makes a real multi-leg
/// handshake possible — and it is the seam the fixtures drive, so every connector test
/// exercises the same loop the real providers do.
pub trait TokenProvider: Send {
    /// Advance the context by one leg.
    fn step(&mut self, peer: Option<&[u8]>) -> StepResult;
}

/// Mints a fresh [`TokenProvider`] per TCP connection.
pub trait ProviderFactory: Send + Sync + 'static {
    /// A new security context for `spn`.
    fn new_provider(&self, spn: &str) -> Result<Box<dyn TokenProvider>, NegotiateError>;

    /// A short name for diagnostics (`sspi`, `gssapi`, `fake`).
    fn name(&self) -> &'static str;
}

/// Why a Negotiate exchange could not proceed.
///
/// The variants are distinct because their remedies are: "install krb5-libs" and "run
/// kinit" are different instructions, and conflating them sends the operator to the wrong
/// one. Every variant carries its own `OL-` code.
#[derive(Debug, Clone)]
pub enum NegotiateError {
    /// No GSSAPI library could be loaded. A degradation, never a load failure: the binary
    /// still starts and still says why.
    LibraryUnavailable(String),
    /// The library is there; there is no usable Kerberos credential.
    NoTicket(String),
    /// The local provider selected NTLM inside SPNEGO. Refused (D-2) — and refused *before*
    /// a token is released, so nothing NTLM-shaped ever reaches the wire.
    NtlmSelected(String),
    /// Anything else the provider reported.
    Provider(String),
}

impl NegotiateError {
    /// The product error this becomes at the call site.
    pub fn into_ol(self) -> OlError {
        match self {
            Self::LibraryUnavailable(detail) => OlError::new(
                ERR_NEGOTIATE_NO_TICKET,
                format!("no GSSAPI library could be loaded: {detail}"),
            )
            .with_suggestion(
                "Install the Kerberos runtime (RHEL/Fedora: krb5-libs; Debian/Ubuntu: \
                 libgssapi-krb5-2), or set [proxy] auth = \"basic\".",
            ),
            Self::NoTicket(detail) => OlError::new(
                ERR_NEGOTIATE_NO_TICKET,
                format!("no Kerberos credential is available: {detail}"),
            )
            .with_suggestion(
                "Obtain a ticket with kinit, or set [proxy] auth = \"basic\". A service \
                 running as a system account usually cannot see a user's ticket cache.",
            ),
            Self::NtlmSelected(detail) => OlError::new(
                ERR_PROXY_SCHEME_UNSUPPORTED,
                format!("the Negotiate provider selected NTLM, which is refused: {detail}"),
            )
            .with_suggestion(
                "NTLM is deprecated and never used. Join the host to the domain so Kerberos \
                 is available, or set [proxy] auth = \"basic\".",
            ),
            Self::Provider(detail) => {
                OlError::new(ERR_PROXY_AUTH_FAILED, format!("Negotiate failed: {detail}"))
                    .with_suggestion(
                        "Check the proxy SPN ([proxy] spn) and that this host holds a valid \
                         Kerberos ticket.",
                    )
            }
        }
    }
}

impl std::fmt::Display for NegotiateError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::LibraryUnavailable(d) => write!(f, "GSSAPI library unavailable: {d}"),
            Self::NoTicket(d) => write!(f, "no Kerberos credential: {d}"),
            Self::NtlmSelected(d) => write!(f, "NTLM selected and refused: {d}"),
            Self::Provider(d) => write!(f, "{d}"),
        }
    }
}

/// The provider this build and this platform can offer, if any.
///
/// A build without a compiled-in provider is a *reportable* state, not a panic: the caller
/// turns it into `OL-1223` with the remedy that names the feature flag.
pub fn platform_provider() -> Result<Arc<dyn ProviderFactory>, NegotiateError> {
    #[cfg(windows)]
    {
        Ok(Arc::new(sspi::SspiProvider::new()))
    }
    #[cfg(unix)]
    {
        Ok(Arc::new(gssapi::GssapiProvider::new()))
    }
    #[cfg(not(any(windows, unix)))]
    {
        Err(NegotiateError::LibraryUnavailable(
            "no SPNEGO provider exists for this platform".to_string(),
        ))
    }
}

// ---------------------------------------------------------------------------
// Streams
// ---------------------------------------------------------------------------

/// The transport to whatever we dialled: the proxy, or the target on a bypass.
pub enum Hop {
    /// Plain TCP.
    Plain(TcpStream),
    /// TLS — an `https://` proxy hop.
    Tls(Box<tokio_rustls::client::TlsStream<TcpStream>>),
}

/// The stream handed up to the HTTP client.
///
/// Two layers, not four variants: [`Hop`] is what we dialled, and the outer `Tls` is the
/// target's own TLS running *inside* whatever that was. A proxied plain-HTTP target is
/// `Bare`, and that is a legal, tested shape — a development `api_url` on `http://` and the
/// mock upstreams both take it.
pub enum NegotiateStream {
    /// No target TLS: a plain-HTTP destination, tunnelled or direct.
    Bare(Hop),
    /// The target's TLS, inside the hop.
    Tls(Box<tokio_rustls::client::TlsStream<Hop>>),
}

/// Delegate `AsyncRead`/`AsyncWrite` to whichever variant is live.
///
/// A macro rather than a `Box<dyn AsyncRead + AsyncWrite>` because the variants are known
/// at compile time and a trait object here would put a vtable dispatch on every byte of
/// every streamed response body.
macro_rules! delegate_io {
    ($ty:ty { $($variant:pat => $inner:expr),+ $(,)? }) => {
        impl AsyncRead for $ty {
            fn poll_read(
                self: Pin<&mut Self>,
                cx: &mut Context<'_>,
                buf: &mut ReadBuf<'_>,
            ) -> Poll<std::io::Result<()>> {
                match self.get_mut() {
                    $($variant => Pin::new($inner).poll_read(cx, buf),)+
                }
            }
        }

        impl AsyncWrite for $ty {
            fn poll_write(
                self: Pin<&mut Self>,
                cx: &mut Context<'_>,
                buf: &[u8],
            ) -> Poll<std::io::Result<usize>> {
                match self.get_mut() {
                    $($variant => Pin::new($inner).poll_write(cx, buf),)+
                }
            }

            fn poll_flush(
                self: Pin<&mut Self>,
                cx: &mut Context<'_>,
            ) -> Poll<std::io::Result<()>> {
                match self.get_mut() {
                    $($variant => Pin::new($inner).poll_flush(cx),)+
                }
            }

            fn poll_shutdown(
                self: Pin<&mut Self>,
                cx: &mut Context<'_>,
            ) -> Poll<std::io::Result<()>> {
                match self.get_mut() {
                    $($variant => Pin::new($inner).poll_shutdown(cx),)+
                }
            }
        }
    };
}

delegate_io!(Hop {
    Hop::Plain(s) => s,
    Hop::Tls(s) => s.as_mut(),
});

delegate_io!(NegotiateStream {
    NegotiateStream::Bare(s) => s,
    NegotiateStream::Tls(s) => s.as_mut(),
});

/// [`NegotiateStream`] wearing the traits hyper's legacy client asks of a connector's
/// output.
///
/// `Connected::proxy(false)` on every variant, including the tunnelled ones: the CONNECT is
/// already established by the time this exists, so hyper must write **origin-form** request
/// lines into it. Reporting `proxy(true)` would make it write absolute-form URIs into an
/// already-open tunnel, which the origin server would reject.
pub struct NegotiateIo {
    inner: hyper_util::rt::TokioIo<NegotiateStream>,
    /// Whether the target's TLS negotiated h2 via ALPN. `Connected` is not publicly
    /// cloneable, so the two facts that define it are kept instead and it is rebuilt on
    /// demand -- there is exactly one call, at the moment hyper adopts the connection.
    negotiated_h2: bool,
}

/// Opaque: nothing inside a live transport is safe to render, and a `Debug` that walked
/// into the TLS session would print session keys.
impl std::fmt::Debug for NegotiateIo {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NegotiateIo")
            .field("negotiated_h2", &self.negotiated_h2)
            .finish_non_exhaustive()
    }
}

/// The route and the provider, with the URL masked. Never the credential ladder.
impl std::fmt::Debug for NegotiateConnector {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NegotiateConnector")
            .field("proxy", &mask_userinfo(&self.inner.proxy_url))
            .field("provider", &self.inner.provider.name())
            .field("spn", &self.spn())
            .finish_non_exhaustive()
    }
}

impl NegotiateIo {
    /// The underlying stream, for a caller that speaks tokio's IO traits rather than
    /// hyper's — the CONNECT-tunnel suites drive raw bytes through it.
    pub fn into_stream(self) -> NegotiateStream {
        self.inner.into_inner()
    }
}

impl hyper::rt::Read for NegotiateIo {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: hyper::rt::ReadBufCursor<'_>,
    ) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.inner).poll_read(cx, buf)
    }
}

impl hyper::rt::Write for NegotiateIo {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        Pin::new(&mut self.inner).poll_write(cx, buf)
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.inner).poll_flush(cx)
    }

    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.inner).poll_shutdown(cx)
    }
}

impl hyper_util::client::legacy::connect::Connection for NegotiateIo {
    fn connected(&self) -> hyper_util::client::legacy::connect::Connected {
        let connected = hyper_util::client::legacy::connect::Connected::new().proxy(false);
        if self.negotiated_h2 {
            connected.negotiated_h2()
        } else {
            connected
        }
    }
}

// ---------------------------------------------------------------------------
// Connector
// ---------------------------------------------------------------------------

/// A CONNECT tunnel that authenticates with SPNEGO.
#[derive(Clone)]
pub struct NegotiateConnector {
    inner: Arc<Inner>,
}

struct Inner {
    /// `scheme://host:port` of the proxy. Never carries userinfo.
    proxy_url: String,
    proxy_scheme: String,
    proxy_host: String,
    proxy_port: u16,
    /// Consulted before the proxy is dialled — see step 0.
    no_proxy: NoProxyMatcher,
    spn_override: Option<String>,
    http1_only: bool,
    connect_timeout: Option<Duration>,
    provider: Arc<dyn ProviderFactory>,
    tls: TlsSetup,
}

impl NegotiateConnector {
    /// Build a connector for `cfg`, using `provider` to mint tokens.
    ///
    /// Fails only on inputs that can never work: no proxy URL, an unsupported proxy scheme,
    /// an unreadable `ca_bundle`. A *network* state is never a construction failure.
    pub fn new(cfg: &EgressConfig, provider: Arc<dyn ProviderFactory>) -> Result<Self, OlError> {
        let Some(url) = cfg.url.as_deref() else {
            return Err(OlError::new(
                ERR_PROXY_SCHEME_UNSUPPORTED,
                "the Negotiate transport needs a proxy url",
            )
            .with_suggestion("Set [proxy] url, or use [proxy] auth = \"none\"."));
        };
        let (scheme, host, port) = split_proxy_url(url)?;
        if !matches!(scheme.as_str(), "http" | "https") {
            // SOCKS carries no HTTP challenge, so there is nothing for SPNEGO to answer.
            return Err(OlError::new(
                ERR_PROXY_SCHEME_UNSUPPORTED,
                format!(
                    "[proxy] auth = negotiate needs an http or https proxy, not \"{scheme}\" \
                     ({})",
                    mask_userinfo(url)
                ),
            )
            .with_suggestion(
                "A SOCKS proxy authenticates with its own username/password exchange. Set \
                 [proxy] auth = \"basic\" or point at an HTTP proxy.",
            ));
        }

        Ok(Self {
            inner: Arc::new(Inner {
                proxy_url: url.to_string(),
                proxy_scheme: scheme,
                proxy_host: host,
                proxy_port: port,
                no_proxy: cfg.no_proxy.clone(),
                spn_override: cfg.spn.clone(),
                http1_only: cfg.http1_only,
                connect_timeout: Some(Duration::from_secs(10)),
                provider,
                tls: TlsSetup::new(cfg)?,
            }),
        })
    }

    /// The SPN this connector will ask for: the override, else `HTTP/<proxy-host>`.
    pub fn spn(&self) -> String {
        self.inner
            .spn_override
            .clone()
            .unwrap_or_else(|| format!("HTTP/{}", self.inner.proxy_host))
    }

    /// Open a transport to `dst`.
    pub async fn connect(&self, dst: http::Uri) -> Result<NegotiateIo, OlError> {
        let inner = self.inner.clone();
        let (host, port, is_tls) = target_of(&dst)?;

        // Step 0. The bypass decision comes FIRST, before the proxy is dialled — see the
        // module docs. reqwest makes this decision inside a closure that has no counterpart
        // on this transport, so without it the two transports would disagree about what
        // bypasses, and loopback would start being proxied.
        if inner.no_proxy.matches(&host, port) {
            let stream = dial(&host, port, inner.connect_timeout).await?;
            let hop = Hop::Plain(stream);
            return inner.finish(hop, &host, is_tls).await;
        }

        // Steps 1-2. The proxy hop, TLS first if the proxy speaks it. Nothing
        // credential-bearing is written before this handshake completes.
        let stream = dial(&inner.proxy_host, inner.proxy_port, inner.connect_timeout).await?;
        let mut hop = if inner.proxy_scheme == "https" {
            Hop::Tls(Box::new(
                inner
                    .tls
                    .connect_proxy(&inner.proxy_host, stream)
                    .await
                    .map_err(|e| {
                        OlError::new(
                            ERR_EGRESS_TLS_FAILED,
                            format!(
                                "TLS to the proxy {} failed: {e}",
                                mask_userinfo(&inner.proxy_url)
                            ),
                        )
                        .with_suggestion(
                            "The proxy's certificate must chain to a trusted root. Add the \
                             interception CA with [proxy] ca_bundle, or install it in the OS \
                             trust store.",
                        )
                    })?,
            ))
        } else {
            Hop::Plain(stream)
        };

        // Steps 3-5. The SPNEGO exchange, on this one connection.
        inner.establish_tunnel(&mut hop, &host, port).await?;

        // Step 6.
        inner.finish(hop, &host, is_tls).await
    }
}

impl Inner {
    /// Wrap the hop in the target's own TLS when the destination is `https`, and hand it up.
    async fn finish(&self, hop: Hop, host: &str, is_tls: bool) -> Result<NegotiateIo, OlError> {
        if !is_tls {
            // A plain-HTTP destination through the tunnel: no target TLS, bytes verbatim.
            return Ok(NegotiateIo {
                inner: hyper_util::rt::TokioIo::new(NegotiateStream::Bare(hop)),
                negotiated_h2: false,
            });
        }

        let stream = self
            .tls
            .connect_target(host, hop, self.http1_only)
            .await
            .map_err(|e| {
                OlError::new(ERR_EGRESS_TLS_FAILED, format!("TLS to {host} failed: {e}"))
                    .with_suggestion(
                        "If a TLS-inspecting proxy is in the path, its CA must be trusted: add \
                     it with [proxy] ca_bundle or install it in the OS trust store.",
                    )
            })?;

        // ALPN is restored on the target hop (it was excluded on the proxy hop), so h2 is
        // reachable for a destination that offers it — but only when the deployment has not
        // pinned HTTP/1.1 for an inspection proxy that mishandles h2.
        let negotiated_h2 = stream.get_ref().1.alpn_protocol() == Some(b"h2");
        Ok(NegotiateIo {
            inner: hyper_util::rt::TokioIo::new(NegotiateStream::Tls(Box::new(stream))),
            negotiated_h2,
        })
    }

    /// Drive the CONNECT legs until the tunnel is open, or the exchange concludes against us.
    async fn establish_tunnel(&self, hop: &mut Hop, host: &str, port: u16) -> Result<(), OlError> {
        let spn = self
            .spn_override
            .clone()
            .unwrap_or_else(|| format!("HTTP/{}", self.proxy_host));

        // Fresh per TCP connection. Squid and MIT both keep a replay cache, so a token
        // reused across connections is rejected — which is why this is here and not cached
        // on the connector.
        let mut provider = self
            .provider
            .new_provider(&spn)
            .map_err(NegotiateError::into_ol)?;

        let authority = format!("{host}:{port}");
        let mut peer: Option<Vec<u8>> = None;

        for leg in 1..=MAX_LEGS {
            let (token, last) = match provider.step(peer.as_deref()) {
                StepResult::Continue(token) => (token, false),
                StepResult::Done(Some(token)) => (token, true),
                StepResult::Done(None) => {
                    // The provider is finished and has nothing left to send, but the proxy
                    // has not accepted. That is a concluded rejection (D-5).
                    return Err(auth_failed(
                        &self.proxy_url,
                        "the security context completed without the proxy accepting it",
                    ));
                }
                StepResult::Failed(e) => return Err(e.into_ol()),
            };

            write_connect(hop, &authority, &token).await?;
            let head = read_head(hop).await?;
            let status = status_of(&head)?;

            match status {
                200 => {
                    // A final token on the 200 is the server's half of mutual auth. It has
                    // to reach the provider BEFORE the tunnel is handed up: after that the
                    // stream carries application bytes and the context can never be
                    // completed.
                    if let Some(token) = negotiate_challenge(&head) {
                        if let StepResult::Failed(e) = provider.step(Some(&token)) {
                            return Err(e.into_ol());
                        }
                    }
                    tracing::debug!(
                        proxy = %mask_userinfo(&self.proxy_url),
                        provider = self.provider.name(),
                        legs = leg,
                        "Negotiate tunnel established"
                    );
                    return Ok(());
                }
                407 => match negotiate_challenge(&head) {
                    // A continuation, on this same connection. Not a retry, and not the
                    // rejection D-5 calls terminal: it is the middle of one handshake.
                    Some(token) if !last => {
                        // Squid and friends put an HTML explanation on a 407. Those bytes
                        // are still in the socket, and the next leg reads a response head
                        // from it -- so an undrained body would be parsed as the proxy's
                        // answer to the next CONNECT. A body whose length we cannot know
                        // (chunked, or connection-delimited) cannot be drained safely, so
                        // the exchange concludes there rather than continuing over bytes
                        // it does not understand.
                        match body_length(&head) {
                            Some(0) => {}
                            Some(len) => drain(hop, len).await?,
                            None => return Err(rejected(&self.proxy_url, &head)),
                        }
                        peer = Some(token);
                        continue;
                    }
                    // Everything else is a concluded rejection: a bare 407, a challenge
                    // this build cannot use, or a 407 after our final token.
                    _ => return Err(rejected(&self.proxy_url, &head)),
                },
                other => {
                    return Err(OlError::new(
                        ERR_PROXY_UNREACHABLE,
                        format!(
                            "the proxy {} answered CONNECT {authority} with {other}",
                            mask_userinfo(&self.proxy_url)
                        ),
                    )
                    .with_suggestion(
                        "The proxy refused the tunnel. Check that it permits CONNECT to this \
                         destination and port.",
                    ))
                }
            }
        }

        Err(auth_failed(
            &self.proxy_url,
            &format!("the SPNEGO exchange did not conclude within {MAX_LEGS} legs"),
        ))
    }
}

impl tower_service::Service<http::Uri> for NegotiateConnector {
    type Response = NegotiateIo;
    type Error = OlError;
    type Future = Pin<Box<dyn Future<Output = Result<NegotiateIo, OlError>> + Send>>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, dst: http::Uri) -> Self::Future {
        let this = self.clone();
        Box::pin(async move { this.connect(dst).await })
    }
}

/// An HTTP client whose every connection is an authenticated Negotiate tunnel.
///
/// The facade's second variant. It is a `hyper_util` legacy client rather than a reqwest
/// one for the reason at the top of this module: reqwest cannot host the connector.
pub fn client(
    connector: NegotiateConnector,
) -> hyper_util::client::legacy::Client<NegotiateConnector, http_body_util::Full<bytes::Bytes>> {
    let mut builder =
        hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new());
    builder.pool_max_idle_per_host(8);
    builder.build(connector)
}

// ---------------------------------------------------------------------------
// Wire helpers
// ---------------------------------------------------------------------------

/// `scheme://host:port` → `(scheme, host, port)`, with the scheme's default port filled in.
fn split_proxy_url(url: &str) -> Result<(String, String, u16), OlError> {
    let bad = || {
        OlError::new(
            ERR_PROXY_SCHEME_UNSUPPORTED,
            format!("proxy url \"{}\" is not usable", mask_userinfo(url)),
        )
        .with_suggestion("Use http://host:port or https://host:port.")
    };
    let (scheme, rest) = url.split_once("://").ok_or_else(bad)?;
    let authority = rest.split('/').next().unwrap_or(rest);
    // Userinfo never reaches here (the parser strips it), but splitting it off costs one
    // line and means a hand-built connector cannot leak one into an SPN.
    let authority = authority.rsplit_once('@').map_or(authority, |(_, h)| h);
    let scheme = scheme.to_ascii_lowercase();
    let (host, port) = split_host_port(authority);
    if host.is_empty() {
        return Err(bad());
    }
    let port = match port {
        Some(p) => p.parse::<u16>().map_err(|_| bad())?,
        None if scheme == "https" => 443,
        None if scheme == "http" => 80,
        None => return Err(bad()),
    };
    Ok((scheme, host.to_string(), port))
}

fn split_host_port(authority: &str) -> (&str, Option<&str>) {
    if let Some(rest) = authority.strip_prefix('[') {
        if let Some((host, tail)) = rest.split_once(']') {
            return (host, tail.strip_prefix(':'));
        }
    }
    match authority.rsplit_once(':') {
        Some((h, p)) => (h, Some(p)),
        None => (authority, None),
    }
}

/// `(host, port, needs_tls)` for a destination URI.
fn target_of(dst: &http::Uri) -> Result<(String, u16, bool), OlError> {
    let scheme = dst.scheme_str().unwrap_or("http").to_ascii_lowercase();
    let is_tls = scheme == "https";
    let host = dst
        .host()
        .ok_or_else(|| {
            OlError::new(
                ERR_PROXY_UNREACHABLE,
                format!("destination \"{dst}\" has no host"),
            )
        })?
        .trim_start_matches('[')
        .trim_end_matches(']')
        .to_string();
    let port = dst.port_u16().unwrap_or(if is_tls { 443 } else { 80 });
    Ok((host, port, is_tls))
}

async fn dial(host: &str, port: u16, timeout: Option<Duration>) -> Result<TcpStream, OlError> {
    let unreachable = |e: std::io::Error| {
        OlError::new(
            ERR_PROXY_UNREACHABLE,
            format!("could not connect to {host}:{port}: {e}"),
        )
        .with_suggestion("Check the proxy host and port, and that this host can reach it.")
    };
    let connect = TcpStream::connect((host, port));
    let stream = match timeout {
        Some(d) => tokio::time::timeout(d, connect).await.map_err(|_| {
            OlError::new(
                ERR_PROXY_UNREACHABLE,
                format!("connecting to {host}:{port} timed out after {d:?}"),
            )
            .with_suggestion("Check the proxy host and port, and that this host can reach it.")
        })?,
        None => connect.await,
    };
    let stream = stream.map_err(unreachable)?;
    // A CONNECT exchange is a handful of small writes; Nagle would add a round trip to
    // every one of them.
    let _ = stream.set_nodelay(true);
    Ok(stream)
}

async fn write_connect<S>(stream: &mut S, authority: &str, token: &[u8]) -> Result<(), OlError>
where
    S: AsyncWrite + Unpin,
{
    let encoded = b64_encode(token);
    let request = format!(
        "CONNECT {authority} HTTP/1.1\r\n\
         Host: {authority}\r\n\
         Proxy-Authorization: Negotiate {encoded}\r\n\
         Proxy-Connection: Keep-Alive\r\n\
         \r\n"
    );
    stream
        .write_all(request.as_bytes())
        .await
        .map_err(|e| write_failed(&e))?;
    stream.flush().await.map_err(|e| write_failed(&e))
}

fn write_failed(e: &std::io::Error) -> OlError {
    OlError::new(
        ERR_PROXY_UNREACHABLE,
        format!("writing CONNECT to the proxy failed: {e}"),
    )
}

/// Read the response head, stopping at the blank line, capped at [`MAX_RESPONSE_HEAD`].
///
/// Byte at a time rather than buffered: the bytes after the head belong to the tunnel, and
/// a buffered reader would swallow them into a buffer this function then has to hand back.
/// A CONNECT response is a few hundred bytes, so the cost is not worth the bug.
async fn read_head<S>(stream: &mut S) -> Result<Vec<u8>, OlError>
where
    S: AsyncRead + Unpin,
{
    use tokio::io::AsyncReadExt;

    let mut head = Vec::with_capacity(256);
    let mut byte = [0u8; 1];
    loop {
        let n = stream.read(&mut byte).await.map_err(|e| {
            OlError::new(
                ERR_PROXY_UNREACHABLE,
                format!("reading the proxy's CONNECT response failed: {e}"),
            )
        })?;
        if n == 0 {
            return Err(OlError::new(
                ERR_PROXY_UNREACHABLE,
                "the proxy closed the connection before answering CONNECT",
            )
            .with_suggestion(
                "The proxy may not permit CONNECT to this destination, or may require a \
                 scheme this build does not speak.",
            ));
        }
        head.push(byte[0]);
        if head.ends_with(b"\r\n\r\n") || head.ends_with(b"\n\n") {
            return Ok(head);
        }
        if head.len() >= MAX_RESPONSE_HEAD {
            return Err(OlError::new(
                ERR_PROXY_UNREACHABLE,
                format!("the proxy's CONNECT response head exceeded {MAX_RESPONSE_HEAD} bytes"),
            ));
        }
    }
}

/// How many body bytes follow this response head, if that is knowable.
///
/// `Some(n)` for an explicit `Content-Length`, `Some(0)` when there is neither a length nor
/// a transfer encoding (the ordinary CONNECT-response shape), and `None` for a chunked or
/// connection-delimited body — which cannot be drained without a full HTTP/1.1 body reader
/// and is therefore treated as the end of the exchange.
fn body_length(head: &[u8]) -> Option<u64> {
    if !header_values(head, "transfer-encoding").is_empty() {
        return None;
    }
    match header_values(head, "content-length").first() {
        Some(value) => value.trim().parse::<u64>().ok(),
        None => Some(0),
    }
}

/// Read and discard `len` bytes, so the next leg reads a response head and not a body.
async fn drain<S>(stream: &mut S, len: u64) -> Result<(), OlError>
where
    S: AsyncRead + Unpin,
{
    use tokio::io::AsyncReadExt;

    // A 407 explanation page is kilobytes. Anything larger is not an explanation, and
    // reading it would be an unbounded read on a peer that has already refused us.
    if len > MAX_RESPONSE_HEAD as u64 {
        return Err(OlError::new(
            ERR_PROXY_UNREACHABLE,
            format!("the proxy's 407 carried a {len}-byte body, which is not an explanation"),
        ));
    }
    let mut sink = vec![0u8; len as usize];
    stream.read_exact(&mut sink).await.map_err(|e| {
        OlError::new(
            ERR_PROXY_UNREACHABLE,
            format!("reading the proxy's 407 body failed: {e}"),
        )
    })?;
    Ok(())
}

/// The status code out of a response head.
fn status_of(head: &[u8]) -> Result<u16, OlError> {
    let mut headers = [httparse::EMPTY_HEADER; 32];
    let mut response = httparse::Response::new(&mut headers);
    match response.parse(head) {
        Ok(_) => response.code.ok_or_else(malformed),
        Err(_) => Err(malformed()),
    }
}

fn malformed() -> OlError {
    OlError::new(
        ERR_PROXY_UNREACHABLE,
        "the proxy's answer to CONNECT was not a valid HTTP response",
    )
    .with_suggestion("Check that [proxy] url points at an HTTP proxy and not at another service.")
}

/// The base64 token from a `Proxy-Authenticate: Negotiate <token>` header, if there is one.
///
/// A bare `Proxy-Authenticate: Negotiate` with no token is `None`: there is nothing to feed
/// the provider, so the exchange has concluded rather than continued.
fn negotiate_challenge(head: &[u8]) -> Option<Vec<u8>> {
    for line in header_values(head, "proxy-authenticate") {
        let (scheme, rest) = match line.split_once(' ') {
            Some(parts) => parts,
            None => continue,
        };
        if !scheme.eq_ignore_ascii_case("Negotiate") {
            continue;
        }
        let token = rest.trim();
        if token.is_empty() {
            continue;
        }
        if let Some(decoded) = b64_decode(token) {
            return Some(decoded);
        }
    }
    None
}

/// Every value of one header name, in order. `Proxy-Authenticate` legitimately repeats.
fn header_values(head: &[u8], name: &str) -> Vec<String> {
    String::from_utf8_lossy(head)
        .lines()
        .skip(1)
        .filter_map(|line| {
            let (k, v) = line.split_once(':')?;
            k.trim()
                .eq_ignore_ascii_case(name)
                .then(|| v.trim().to_string())
        })
        .collect()
}

/// The schemes a 407 offered, for the error text.
fn offered_schemes(head: &[u8]) -> Vec<String> {
    header_values(head, "proxy-authenticate")
        .into_iter()
        .filter_map(|v| v.split_whitespace().next().map(|s| s.to_ascii_lowercase()))
        .collect()
}

/// A concluded 407. Which code it carries depends on *why* the proxy said no.
fn rejected(proxy_url: &str, head: &[u8]) -> OlError {
    let schemes = offered_schemes(head);
    let masked = mask_userinfo(proxy_url);

    // A proxy that offers only schemes we refuse (NTLM) or do not implement is a
    // *scheme* problem, not a credential problem, and the remedy is different.
    let negotiable = schemes.iter().any(|s| s == "negotiate");
    if !schemes.is_empty() && !negotiable {
        return OlError::new(
            ERR_PROXY_SCHEME_UNSUPPORTED,
            format!(
                "the proxy {masked} offers only {} — Negotiate is not among them",
                schemes.join(", ")
            ),
        )
        .with_suggestion(
            "NTLM is deliberately not supported (deprecated by Microsoft in 2024). Set \
             [proxy] auth = \"basic\" if the proxy also offers Basic, or ask for Kerberos \
             to be enabled on it.",
        );
    }

    auth_failed(proxy_url, "the proxy rejected the Kerberos credential")
}

fn auth_failed(proxy_url: &str, detail: &str) -> OlError {
    OlError::new(
        ERR_PROXY_AUTH_FAILED,
        format!("Negotiate to {}: {detail}", mask_userinfo(proxy_url)),
    )
    .with_suggestion(
        "Check that this host holds a valid Kerberos ticket (klist) and that [proxy] spn \
         matches the proxy's service principal.",
    )
}

const B64: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";

fn b64_encode(input: &[u8]) -> String {
    let mut out = String::with_capacity(input.len().div_ceil(3) * 4);
    for chunk in input.chunks(3) {
        let a = chunk[0] as u32;
        let b = *chunk.get(1).unwrap_or(&0) as u32;
        let c = *chunk.get(2).unwrap_or(&0) as u32;
        let packed = (a << 16) | (b << 8) | c;
        out.push(B64[((packed >> 18) & 63) as usize] as char);
        out.push(B64[((packed >> 12) & 63) as usize] as char);
        out.push(if chunk.len() > 1 {
            B64[((packed >> 6) & 63) as usize] as char
        } else {
            '='
        });
        out.push(if chunk.len() > 2 {
            B64[(packed & 63) as usize] as char
        } else {
            '='
        });
    }
    out
}

fn b64_decode(input: &str) -> Option<Vec<u8>> {
    let mut acc: u32 = 0;
    let mut bits: u8 = 0;
    let mut out = Vec::new();
    for byte in input.bytes() {
        if matches!(byte, b'=' | b'\r' | b'\n' | b' ' | b'\t') {
            continue;
        }
        let value = B64.iter().position(|&b| b == byte)? as u32;
        acc = (acc << 6) | value;
        bits += 6;
        if bits >= 8 {
            bits -= 8;
            out.push((acc >> bits) as u8);
        }
    }
    Some(out)
}

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

    #[test]
    fn base64_round_trips() {
        for probe in [
            &b""[..],
            b"a",
            b"ab",
            b"abc",
            b"\x60\x82\x01\x0c\x06\x06\x2b\x06\x01\x05\x05\x02",
        ] {
            let encoded = b64_encode(probe);
            assert_eq!(b64_decode(&encoded).as_deref(), Some(probe));
        }
    }

    #[test]
    fn the_proxy_url_splits_with_the_scheme_default_port() {
        assert_eq!(
            split_proxy_url("http://proxy.corp").expect("split"),
            ("http".into(), "proxy.corp".into(), 80)
        );
        assert_eq!(
            split_proxy_url("https://proxy.corp").expect("split"),
            ("https".into(), "proxy.corp".into(), 443)
        );
        assert_eq!(
            split_proxy_url("http://proxy.corp:3128").expect("split"),
            ("http".into(), "proxy.corp".into(), 3128)
        );
        assert!(split_proxy_url("http://").is_err());
        assert!(split_proxy_url("proxy.corp:3128").is_err());
    }

    #[test]
    fn the_target_defaults_its_port_from_the_scheme() {
        let https: http::Uri = "https://api.openlatch.ai/v1".parse().expect("uri");
        assert_eq!(
            target_of(&https).expect("target"),
            ("api.openlatch.ai".to_string(), 443, true)
        );
        // A plain-http destination is legal -- a development api_url takes exactly this
        // shape -- and must NOT be given target TLS.
        let http: http::Uri = "http://upstream.test/hello".parse().expect("uri");
        assert_eq!(
            target_of(&http).expect("target"),
            ("upstream.test".to_string(), 80, false)
        );
        let explicit: http::Uri = "http://upstream.test:8080/".parse().expect("uri");
        assert_eq!(
            target_of(&explicit).expect("target"),
            ("upstream.test".to_string(), 8080, false)
        );
    }

    #[test]
    fn a_challenge_token_is_decoded_only_for_negotiate() {
        let head = b"HTTP/1.1 407 Proxy Authentication Required\r\n\
                     Proxy-Authenticate: NTLM\r\n\
                     Proxy-Authenticate: Negotiate YWJj\r\n\r\n";
        assert_eq!(negotiate_challenge(head).as_deref(), Some(&b"abc"[..]));
        assert_eq!(offered_schemes(head), vec!["ntlm", "negotiate"]);
    }

    /// A bare `Negotiate` with no token has nothing to feed the provider, so the exchange
    /// has concluded rather than continued -- which is what makes it terminal.
    #[test]
    fn a_bare_negotiate_challenge_carries_no_token() {
        let head = b"HTTP/1.1 407 Proxy Authentication Required\r\n\
                     Proxy-Authenticate: Negotiate\r\n\r\n";
        assert_eq!(negotiate_challenge(head), None);
        assert_eq!(offered_schemes(head), vec!["negotiate"]);
    }

    /// D-2 on the wire: an NTLM-only proxy is a *scheme* refusal (OL-1223), never a
    /// credential retry, and the remedy says so.
    #[test]
    fn an_ntlm_only_challenge_is_a_scheme_refusal() {
        let head = b"HTTP/1.1 407 Proxy Authentication Required\r\n\
                     Proxy-Authenticate: NTLM\r\n\r\n";
        let err = rejected("http://proxy.corp:8080", head);
        assert_eq!(err.code, ERR_PROXY_SCHEME_UNSUPPORTED);
        assert!(err.message.contains("ntlm"), "{}", err.message);
    }

    #[test]
    fn a_bare_407_is_an_auth_failure() {
        let head = b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n";
        assert_eq!(
            rejected("http://proxy.corp:8080", head).code,
            ERR_PROXY_AUTH_FAILED
        );
    }

    /// A 407 body left in the socket would be read as the *next* leg's response head. The
    /// length has to be known before the exchange may continue over it.
    #[test]
    fn the_body_length_of_a_challenge_is_known_or_the_exchange_stops() {
        let no_body = b"HTTP/1.1 407 x
Proxy-Authenticate: Negotiate YWJj

";
        assert_eq!(body_length(no_body), Some(0));

        let explained = b"HTTP/1.1 407 x
Content-Length: 42

";
        assert_eq!(body_length(explained), Some(42));

        // Chunked cannot be drained without a full body reader, so it concludes instead.
        let chunked = b"HTTP/1.1 407 x
Transfer-Encoding: chunked

";
        assert_eq!(body_length(chunked), None);
    }

    #[test]
    fn the_status_line_parses_and_garbage_does_not() {
        assert_eq!(
            status_of(b"HTTP/1.1 200 Connection Established\r\n\r\n").expect("status"),
            200
        );
        assert_eq!(
            status_of(b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n").expect("status"),
            407
        );
        assert!(status_of(b"definitely not http\r\n\r\n").is_err());
    }

    /// A password in a proxy URL must never reach an error message, on any of the paths
    /// this module builds one.
    #[test]
    fn no_error_path_renders_a_password() {
        let url = "http://alice:hunter2@proxy.corp:8080";
        let rendered = format!(
            "{:?} {:?} {:?}",
            auth_failed(url, "detail"),
            rejected(url, b"HTTP/1.1 407 x\r\nProxy-Authenticate: NTLM\r\n\r\n"),
            split_proxy_url("alice:hunter2@proxy.corp").expect_err("must fail")
        );
        assert!(
            !rendered.contains("hunter2"),
            "a password leaked into an error: {rendered}"
        );
    }

    #[test]
    fn negotiate_errors_carry_distinct_codes_and_distinct_remedies() {
        // "install krb5-libs" and "run kinit" are different instructions; conflating them
        // sends the operator to the wrong one.
        let unavailable = NegotiateError::LibraryUnavailable("none found".into()).into_ol();
        let no_ticket = NegotiateError::NoTicket("empty cache".into()).into_ol();
        let ntlm = NegotiateError::NtlmSelected("workgroup host".into()).into_ol();
        let other = NegotiateError::Provider("bad SPN".into()).into_ol();

        assert_eq!(unavailable.code, ERR_NEGOTIATE_NO_TICKET);
        assert_eq!(no_ticket.code, ERR_NEGOTIATE_NO_TICKET);
        assert_eq!(ntlm.code, ERR_PROXY_SCHEME_UNSUPPORTED);
        assert_eq!(other.code, ERR_PROXY_AUTH_FAILED);

        assert!(unavailable
            .suggestion
            .as_deref()
            .is_some_and(|s| s.contains("libgssapi-krb5-2")));
        assert!(no_ticket
            .suggestion
            .as_deref()
            .is_some_and(|s| s.contains("kinit")));
        assert!(ntlm.message.contains("NTLM"));
    }

    #[test]
    fn a_socks_proxy_is_refused_rather_than_silently_downgraded() {
        // SOCKS carries no HTTP challenge, so there is nothing for SPNEGO to answer.
        let mut cfg = EgressConfig::direct();
        cfg.url = Some("socks5://proxy.corp:1080".into());
        let err = NegotiateConnector::new(&cfg, Arc::new(NeverProvider))
            .expect_err("socks must be refused");
        assert_eq!(err.code, ERR_PROXY_SCHEME_UNSUPPORTED);
    }

    #[test]
    fn the_spn_defaults_to_the_proxy_host_and_the_override_wins() {
        let mut cfg = EgressConfig::direct();
        cfg.url = Some("http://proxy.corp:8080".into());
        let connector =
            NegotiateConnector::new(&cfg, Arc::new(NeverProvider)).expect("connector builds");
        assert_eq!(connector.spn(), "HTTP/proxy.corp");

        cfg.spn = Some("HTTP/proxy-alias.corp".into());
        let connector =
            NegotiateConnector::new(&cfg, Arc::new(NeverProvider)).expect("connector builds");
        assert_eq!(connector.spn(), "HTTP/proxy-alias.corp");
    }

    /// A factory that refuses, for the construction-only tests above.
    struct NeverProvider;
    impl ProviderFactory for NeverProvider {
        fn new_provider(&self, _spn: &str) -> Result<Box<dyn TokenProvider>, NegotiateError> {
            Err(NegotiateError::NoTicket("test provider".into()))
        }
        fn name(&self) -> &'static str {
            "never"
        }
    }
}