featherbit/outbound/

mod.rs

//! Shared outbound HTTP client for plugin callouts and upstream proxying.
//!
//! One pooled hyper client pair lives in `PluginResources` for the process
//! lifetime: plugins (forward-auth, opa, loggers, upstream, ...) reuse its
//! connection pool instead of constructing a client per node or per request.
//! Supports `http` and `https` (rustls, native roots); `ssl_verify: false`
//! selects a lazily-built client with certificate verification disabled.

pub mod tls;

use std::collections::HashMap;
use std::sync::{Arc, OnceLock};
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use hyper_rustls::HttpsConnector;
use hyper_util::client::legacy::connect::HttpConnector;
use hyper_util::client::legacy::Client;
use hyper_util::rt::TokioExecutor;

type PooledClient = Client<HttpsConnector<HttpConnector>, Full<Bytes>>;

/// A single outbound request. `timeout` covers the whole call: connect,
/// request write, and response body collection.
pub struct OutboundRequest {
    pub method: http::Method,
    pub url: String,
    /// Header name/value pairs; names may repeat for multi-value headers.
    pub headers: Vec<(String, String)>,
    pub body: Bytes,
    /// Whole-call deadline. Callers should default to 3s for callouts.
    pub timeout: Duration,
    /// When false, TLS certificate verification is disabled (matching
    /// APISIX's `ssl_verify: false`). Ignored for plain-http URLs and when
    /// `tls` is set (the identity carries its own verify flag).
    pub ssl_verify: bool,
    /// Per-upstream TLS identity (client cert / private CA). `None` uses the
    /// shared verified/insecure clients — today's behavior.
    pub tls: Option<Arc<tls::UpstreamTls>>,
}

impl OutboundRequest {
    /// A GET with the callout default timeout (3s) and TLS verification on.
    #[allow(dead_code)] // convenience ctor; callers currently build the struct literally
    pub fn new(method: http::Method, url: impl Into<String>) -> Self {
        Self {
            method,
            url: url.into(),
            headers: Vec::new(),
            body: Bytes::new(),
            timeout: Duration::from_secs(3),
            ssl_verify: true,
            tls: None,
        }
    }
}

/// A fully-buffered outbound response.
pub struct OutboundResponse {
    pub status: u16,
    pub headers: HashMap<String, Vec<String>>,
    pub body: Bytes,
}

/// Outbound call failure, distinguishing timeouts from transport errors so
/// callers can map them to distinct gateway error codes.
#[derive(Debug)]
pub enum OutboundError {
    Timeout(Duration),
    /// Request could not be built (bad URL/header) — a config-shaped error.
    InvalidRequest(String),
    /// Connect/transport/body failure.
    Transport(String),
}

impl std::fmt::Display for OutboundError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Timeout(d) => write!(f, "outbound request timed out after {:?}", d),
            Self::InvalidRequest(m) => write!(f, "invalid outbound request: {}", m),
            Self::Transport(m) => write!(f, "outbound transport error: {}", m),
        }
    }
}

/// Process-wide pooled HTTP client.
pub struct OutboundClient {
    verified: PooledClient,
    /// Built on first `ssl_verify: false` request; never constructed on the
    /// common path.
    insecure: OnceLock<PooledClient>,
    /// Clients for per-upstream TLS identities (mTLS / private CA), keyed by
    /// the identity's content hash. Never evicted — bounded by the number of
    /// distinct identities ever configured.
    custom: std::sync::RwLock<HashMap<u64, PooledClient>>,
}

impl OutboundClient {
    /// Builds the shared client. TLS uses the platform's native root store.
    /// Advertises both HTTP/1.1 and HTTP/2 via ALPN, so TLS upstreams that
    /// support h2 (e.g. gRPC backends) are served over HTTP/2 while plain-`http`
    /// upstreams stay HTTP/1.1.
    pub fn new() -> Self {
        // Pin the process-level rustls provider to ring (both backends compile
        // in through the dependency tree). Shared with the listeners.
        crate::server::tls::install_crypto_provider();

        let https = hyper_rustls::HttpsConnectorBuilder::new()
            .with_native_roots()
            .expect("failed to load native TLS roots")
            .https_or_http()
            .enable_http1()
            .enable_http2()
            .build();
        Self {
            verified: Client::builder(TokioExecutor::new()).build(https),
            insecure: OnceLock::new(),
            custom: std::sync::RwLock::new(HashMap::new()),
        }
    }

    /// Performs the request, honoring `timeout` and `ssl_verify`.
    pub async fn request(&self, req: OutboundRequest) -> Result<OutboundResponse, OutboundError> {
        let mut builder = http::Request::builder().method(req.method).uri(&req.url);
        for (name, value) in &req.headers {
            builder = builder.header(name.as_str(), value.as_str());
        }
        let request = builder
            .body(Full::new(req.body))
            .map_err(|e| OutboundError::InvalidRequest(e.to_string()))?;

        let custom_client;
        let client = match &req.tls {
            Some(identity) => {
                custom_client = self.identity_client(identity)?;
                &custom_client
            }
            None if req.ssl_verify => &self.verified,
            None => self.insecure.get_or_init(build_insecure_client),
        };

        let deadline = req.timeout;
        let call = async {
            let response = client
                .request(request)
                .await
                .map_err(|e| OutboundError::Transport(e.to_string()))?;

            let status = response.status().as_u16();
            let mut headers: HashMap<String, Vec<String>> = HashMap::new();
            for (name, value) in response.headers() {
                headers
                    .entry(name.as_str().to_string())
                    .or_default()
                    .push(value.to_str().unwrap_or("").to_string());
            }
            let body = response
                .into_body()
                .collect()
                .await
                .map_err(|e| OutboundError::Transport(e.to_string()))?
                .to_bytes();

            Ok(OutboundResponse {
                status,
                headers,
                body,
            })
        };

        tokio::time::timeout(deadline, call)
            .await
            .map_err(|_| OutboundError::Timeout(deadline))?
    }

    /// Returns the pooled client for `identity`, building it on first use.
    /// ALPN advertises h1+h2 like the default client (hyper-rustls sets the
    /// protocols on the config in `enable_http1`/`enable_http2`).
    fn identity_client(
        &self,
        identity: &Arc<tls::UpstreamTls>,
    ) -> Result<PooledClient, OutboundError> {
        if let Some(c) = self.custom.read().unwrap().get(&identity.cache_key()) {
            return Ok(c.clone());
        }
        // Materials were validated at policy compile; failure here is
        // config-shaped (e.g. native roots unavailable), not transport.
        let config = identity
            .client_config()
            .map_err(OutboundError::InvalidRequest)?;
        let https = hyper_rustls::HttpsConnectorBuilder::new()
            .with_tls_config(config)
            .https_or_http()
            .enable_http1()
            .enable_http2()
            .build();
        let client = Client::builder(TokioExecutor::new()).build(https);
        // Race-safe: whoever loses uses the winner's client.
        Ok(self
            .custom
            .write()
            .unwrap()
            .entry(identity.cache_key())
            .or_insert(client)
            .clone())
    }
}

impl Default for OutboundClient {
    fn default() -> Self {
        Self::new()
    }
}

fn build_insecure_client() -> PooledClient {
    let config = rustls::ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(NoVerification(
            rustls::crypto::ring::default_provider(),
        )))
        .with_no_client_auth();
    let https = hyper_rustls::HttpsConnectorBuilder::new()
        .with_tls_config(config)
        .https_or_http()
        .enable_http1()
        .build();
    Client::builder(TokioExecutor::new()).build(https)
}

/// Builds a client TLS connector for a raw upstream WebSocket (`wss`) handshake.
///
/// ALPN is pinned to `http/1.1` (the WebSocket upgrade is HTTP/1.1). When
/// `identity` is set, the connector presents that client certificate (and
/// trusts its private CA, if any) — mirroring `OutboundClient::identity_client`
/// — and is cached per identity (`UpstreamTls::cache_key`), never evicted. When
/// `identity` is `None` and `verify` is false, certificate verification is
/// disabled (matching `ssl_verify: false`). The verified connector is cached,
/// since loading the platform's native root store on every connection is
/// wasteful; the insecure connector is cheap and built on demand.
pub fn client_tls_connector(
    verify: bool,
    identity: Option<&Arc<tls::UpstreamTls>>,
) -> Result<tokio_rustls::TlsConnector, String> {
    crate::server::tls::install_crypto_provider();

    if let Some(id) = identity {
        static CUSTOM: OnceLock<std::sync::RwLock<HashMap<u64, tokio_rustls::TlsConnector>>> =
            OnceLock::new();
        let cache = CUSTOM.get_or_init(|| std::sync::RwLock::new(HashMap::new()));
        if let Some(c) = cache.read().unwrap().get(&id.cache_key()) {
            return Ok(c.clone());
        }
        let mut config = id.client_config()?;
        config.alpn_protocols = vec![b"http/1.1".to_vec()];
        let connector = tokio_rustls::TlsConnector::from(Arc::new(config));
        return Ok(cache
            .write()
            .unwrap()
            .entry(id.cache_key())
            .or_insert(connector)
            .clone());
    }

    if !verify {
        let config = rustls::ClientConfig::builder()
            .dangerous()
            .with_custom_certificate_verifier(Arc::new(NoVerification(
                rustls::crypto::ring::default_provider(),
            )))
            .with_no_client_auth();
        let mut config = config;
        config.alpn_protocols = vec![b"http/1.1".to_vec()];
        return Ok(tokio_rustls::TlsConnector::from(Arc::new(config)));
    }

    static VERIFIED: OnceLock<tokio_rustls::TlsConnector> = OnceLock::new();
    if let Some(c) = VERIFIED.get() {
        return Ok(c.clone());
    }

    let mut roots = rustls::RootCertStore::empty();
    let loaded = rustls_native_certs::load_native_certs();
    let (added, _ignored) = roots.add_parsable_certificates(loaded.certs);
    if added == 0 {
        return Err(format!(
            "no usable native root certificates ({} load error(s))",
            loaded.errors.len()
        ));
    }
    let mut config = rustls::ClientConfig::builder()
        .with_root_certificates(roots)
        .with_no_client_auth();
    config.alpn_protocols = vec![b"http/1.1".to_vec()];
    let connector = tokio_rustls::TlsConnector::from(Arc::new(config));
    // Race-safe: whoever loses just uses the winner's connector.
    Ok(VERIFIED.get_or_init(|| connector).clone())
}

/// Certificate verifier that accepts everything — only reachable via an
/// explicit `ssl_verify: false` in plugin config.
#[derive(Debug)]
pub(crate) struct NoVerification(pub(crate) rustls::crypto::CryptoProvider);

impl rustls::client::danger::ServerCertVerifier for NoVerification {
    fn verify_server_cert(
        &self,
        _end_entity: &rustls::pki_types::CertificateDer<'_>,
        _intermediates: &[rustls::pki_types::CertificateDer<'_>],
        _server_name: &rustls::pki_types::ServerName<'_>,
        _ocsp_response: &[u8],
        _now: rustls::pki_types::UnixTime,
    ) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
        Ok(rustls::client::danger::ServerCertVerified::assertion())
    }

    fn verify_tls12_signature(
        &self,
        message: &[u8],
        cert: &rustls::pki_types::CertificateDer<'_>,
        dss: &rustls::DigitallySignedStruct,
    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
        rustls::crypto::verify_tls12_signature(
            message,
            cert,
            dss,
            &self.0.signature_verification_algorithms,
        )
    }

    fn verify_tls13_signature(
        &self,
        message: &[u8],
        cert: &rustls::pki_types::CertificateDer<'_>,
        dss: &rustls::DigitallySignedStruct,
    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
        rustls::crypto::verify_tls13_signature(
            message,
            cert,
            dss,
            &self.0.signature_verification_algorithms,
        )
    }

    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
        self.0.signature_verification_algorithms.supported_schemes()
    }
}

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

    #[test]
    fn test_client_tls_connector_insecure_builds() {
        // Insecure connector never touches the native store — fast/deterministic.
        assert!(client_tls_connector(false, None).is_ok());
    }

    #[test]
    fn test_client_tls_connector_verified_builds() {
        // On a normal dev/CI host the native root store loads; if an
        // environment has no parsable roots the helper returns Err gracefully.
        assert!(client_tls_connector(true, None).is_ok());
    }

    #[test]
    fn test_client_tls_connector_with_identity_builds_and_caches() {
        // Reuse the identity written by the Task 1 helper — write PEMs inline
        // here the same way (CA + leaf via rcgen, temp files).
        let (cert, key, ca) = {
            let mut ca_params = rcgen::CertificateParams::new(Vec::<String>::new()).unwrap();
            ca_params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
            let ca_key = rcgen::KeyPair::generate().unwrap();
            let ca_cert = ca_params.self_signed(&ca_key).unwrap();
            let leaf_params = rcgen::CertificateParams::new(vec!["client".to_string()]).unwrap();
            let leaf_key = rcgen::KeyPair::generate().unwrap();
            let leaf_cert = leaf_params.signed_by(&leaf_key, &ca_cert, &ca_key).unwrap();
            let dir = std::env::temp_dir();
            let pid = std::process::id();
            let cert = dir.join(format!("featherbit_wsid_{}.crt", pid));
            let key = dir.join(format!("featherbit_wsid_{}.key", pid));
            let ca = dir.join(format!("featherbit_wsid_{}.ca.crt", pid));
            std::fs::write(&cert, leaf_cert.pem()).unwrap();
            std::fs::write(&key, leaf_key.serialize_pem()).unwrap();
            std::fs::write(&ca, ca_cert.pem()).unwrap();
            (
                cert.to_str().unwrap().to_string(),
                key.to_str().unwrap().to_string(),
                ca.to_str().unwrap().to_string(),
            )
        };
        let identity = tls::UpstreamTls::load(Some((&cert, &key)), Some(&ca), true).unwrap();
        assert!(client_tls_connector(true, Some(&identity)).is_ok());
        // Second call hits the connector cache — still fine.
        assert!(client_tls_connector(true, Some(&identity)).is_ok());
        // No identity: existing behavior, both variants still build.
        assert!(client_tls_connector(true, None).is_ok());
        assert!(client_tls_connector(false, None).is_ok());
    }

    /// Minimal one-shot HTTPS server that requires a client certificate and
    /// answers any request with `HTTP/1.1 200 OK`. Returns its port.
    async fn spawn_mtls_server(server_config: Arc<rustls::ServerConfig>) -> u16 {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        tokio::spawn(async move {
            let acceptor = tokio_rustls::TlsAcceptor::from(server_config);
            // Serve connections until the test ends; failed handshakes
            // (missing client cert) just drop the connection.
            loop {
                let Ok((tcp, _)) = listener.accept().await else {
                    break;
                };
                let acceptor = acceptor.clone();
                tokio::spawn(async move {
                    if let Ok(mut stream) = acceptor.accept(tcp).await {
                        use tokio::io::{AsyncReadExt, AsyncWriteExt};
                        let mut buf = [0u8; 4096];
                        let _ = stream.read(&mut buf).await;
                        let _ = stream
                            .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\nok")
                            .await;
                        let _ = stream.shutdown().await;
                    }
                });
            }
        });
        port
    }

    #[tokio::test]
    async fn test_request_with_client_identity_reaches_mtls_backend() {
        crate::server::tls::install_crypto_provider();

        // CA, a server cert for "localhost", and a client cert — one CA for both.
        let mut ca_params = rcgen::CertificateParams::new(Vec::<String>::new()).unwrap();
        ca_params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
        let ca_key = rcgen::KeyPair::generate().unwrap();
        let ca_cert = ca_params.self_signed(&ca_key).unwrap();

        let server_params = rcgen::CertificateParams::new(vec!["localhost".to_string()]).unwrap();
        let server_key = rcgen::KeyPair::generate().unwrap();
        let server_cert = server_params
            .signed_by(&server_key, &ca_cert, &ca_key)
            .unwrap();

        let client_params = rcgen::CertificateParams::new(vec!["gw".to_string()]).unwrap();
        let client_key = rcgen::KeyPair::generate().unwrap();
        let client_cert = client_params
            .signed_by(&client_key, &ca_cert, &ca_key)
            .unwrap();

        // Server side: require a client cert signed by the CA.
        let mut roots = rustls::RootCertStore::empty();
        roots
            .add(rustls::pki_types::CertificateDer::from(
                ca_cert.der().to_vec(),
            ))
            .unwrap();
        let verifier = rustls::server::WebPkiClientVerifier::builder(Arc::new(roots))
            .build()
            .unwrap();
        let server_config = rustls::ServerConfig::builder()
            .with_client_cert_verifier(verifier)
            .with_single_cert(
                vec![rustls::pki_types::CertificateDer::from(
                    server_cert.der().to_vec(),
                )],
                rustls::pki_types::PrivateKeyDer::try_from(server_key.serialize_der()).unwrap(),
            )
            .unwrap();
        let port = spawn_mtls_server(Arc::new(server_config)).await;

        // Gateway side: identity = client cert + key, CA bundle for the server.
        let dir = std::env::temp_dir();
        let pid = std::process::id();
        let cert_path = dir.join(format!("featherbit_ob_mtls_{}.crt", pid));
        let key_path = dir.join(format!("featherbit_ob_mtls_{}.key", pid));
        let ca_path = dir.join(format!("featherbit_ob_mtls_{}.ca.crt", pid));
        std::fs::write(&cert_path, client_cert.pem()).unwrap();
        std::fs::write(&key_path, client_key.serialize_pem()).unwrap();
        std::fs::write(&ca_path, ca_cert.pem()).unwrap();

        let identity = tls::UpstreamTls::load(
            Some((cert_path.to_str().unwrap(), key_path.to_str().unwrap())),
            Some(ca_path.to_str().unwrap()),
            true,
        )
        .unwrap();

        let client = OutboundClient::new();
        let ok = client
            .request(OutboundRequest {
                method: http::Method::GET,
                url: format!("https://localhost:{}/", port),
                headers: Vec::new(),
                body: Bytes::new(),
                timeout: Duration::from_secs(5),
                ssl_verify: true,
                tls: Some(identity),
            })
            .await
            .expect("mTLS request should succeed");
        assert_eq!(ok.status, 200);

        // Without a client cert (CA-only identity) the handshake is rejected.
        let ca_only = tls::UpstreamTls::load(None, Some(ca_path.to_str().unwrap()), true).unwrap();
        let err = client
            .request(OutboundRequest {
                method: http::Method::GET,
                url: format!("https://localhost:{}/", port),
                headers: Vec::new(),
                body: Bytes::new(),
                timeout: Duration::from_secs(5),
                ssl_verify: true,
                tls: Some(ca_only),
            })
            .await;
        assert!(
            matches!(err, Err(OutboundError::Transport(_))),
            "got: {:?}",
            err.map(|r| r.status)
        );
    }
}