pingora_core/upstreams/

peer.rs

// Copyright 2026 Cloudflare, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Defines where to connect to and how to connect to a remote server

use crate::connectors::{l4::BindTo, L4Connect};
use crate::protocols::l4::socket::SocketAddr;
use crate::protocols::tls::CaType;
#[cfg(feature = "openssl_derived")]
use crate::protocols::tls::HandshakeCompleteHook;
#[cfg(feature = "s2n")]
use crate::protocols::tls::PskType;
#[cfg(unix)]
use crate::protocols::ConnFdReusable;
use crate::protocols::TcpKeepalive;
use crate::utils::tls::{get_organization_unit, CertKey};
use ahash::AHasher;
use derivative::Derivative;
use pingora_error::{
    ErrorType::{InternalError, SocketError},
    OrErr, Result,
};
#[cfg(feature = "s2n")]
use pingora_s2n::S2NPolicy;
use std::borrow::Cow;
use std::collections::BTreeMap;
use std::fmt::{Display, Formatter, Result as FmtResult};
use std::hash::{Hash, Hasher};
use std::net::{IpAddr, SocketAddr as InetSocketAddr, ToSocketAddrs as ToInetSocketAddrs};
#[cfg(unix)]
use std::os::unix::{net::SocketAddr as UnixSocketAddr, prelude::AsRawFd};
#[cfg(windows)]
use std::os::windows::io::AsRawSocket;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use tokio::net::TcpSocket;

pub use crate::protocols::tls::ALPN;

/// A hook function that may generate user data for [`crate::protocols::raw_connect::ProxyDigest`].
///
/// Takes the request and response headers from the proxy connection establishment, and may produce
/// arbitrary data to be stored in ProxyDigest's user_data field.
///
/// This can be useful when, for example, you want to store some parameter(s) from the request or
/// response headers from when the proxy connection was first established.
pub type ProxyDigestUserDataHook = Arc<
    dyn Fn(
            &http::request::Parts,         // request headers
            &pingora_http::ResponseHeader, // response headers
        ) -> Option<Box<dyn std::any::Any + Send + Sync>>
        + Send
        + Sync
        + 'static,
>;

/// The interface to trace the connection
pub trait Tracing: Send + Sync + std::fmt::Debug {
    /// This method is called when successfully connected to a remote server
    fn on_connected(&self);
    /// This method is called when the connection is disconnected.
    fn on_disconnected(&self);
    /// A way to clone itself
    fn boxed_clone(&self) -> Box<dyn Tracing>;
}

/// An object-safe version of Tracing object that can use Clone
#[derive(Debug)]
pub struct Tracer(pub Box<dyn Tracing>);

impl Clone for Tracer {
    fn clone(&self) -> Self {
        Tracer(self.0.boxed_clone())
    }
}

/// [`Peer`] defines the interface to communicate with the [`crate::connectors`] regarding where to
/// connect to and how to connect to it.
pub trait Peer: Display + Clone {
    /// The remote address to connect to
    fn address(&self) -> &SocketAddr;
    /// If TLS should be used;
    fn tls(&self) -> bool;
    /// The SNI to send, if TLS is used
    fn sni(&self) -> &str;
    /// To decide whether a [`Peer`] can use the connection established by another [`Peer`].
    ///
    /// The connections to two peers are considered reusable to each other if their reuse hashes are
    /// the same
    fn reuse_hash(&self) -> u64;
    /// Get the proxy setting to connect to the remote server
    fn get_proxy(&self) -> Option<&Proxy> {
        None
    }
    /// Get the additional options to connect to the peer.
    ///
    /// See [`PeerOptions`] for more details
    fn get_peer_options(&self) -> Option<&PeerOptions> {
        None
    }
    /// Get the additional options for modification.
    fn get_mut_peer_options(&mut self) -> Option<&mut PeerOptions> {
        None
    }
    /// Whether the TLS handshake should validate the cert of the server.
    fn verify_cert(&self) -> bool {
        match self.get_peer_options() {
            Some(opt) => opt.verify_cert,
            None => false,
        }
    }
    /// Whether the TLS handshake should verify that the server cert matches the SNI.
    fn verify_hostname(&self) -> bool {
        match self.get_peer_options() {
            Some(opt) => opt.verify_hostname,
            None => false,
        }
    }
    /// Whether the system trust store should be loaded and used when verifying certificates
    #[cfg(feature = "s2n")]
    fn use_system_certs(&self) -> bool {
        match self.get_peer_options() {
            Some(opt) => opt.use_system_certs,
            None => false,
        }
    }
    /// The alternative common name to use to verify the server cert.
    ///
    /// If the server cert doesn't match the SNI, this name will be used to
    /// verify the cert.
    fn alternative_cn(&self) -> Option<&String> {
        match self.get_peer_options() {
            Some(opt) => opt.alternative_cn.as_ref(),
            None => None,
        }
    }
    /// Information about the local source address this connection should be bound to.
    fn bind_to(&self) -> Option<&BindTo> {
        match self.get_peer_options() {
            Some(opt) => opt.bind_to.as_ref(),
            None => None,
        }
    }
    /// How long connect() call should be wait before it returns a timeout error.
    fn connection_timeout(&self) -> Option<Duration> {
        match self.get_peer_options() {
            Some(opt) => opt.connection_timeout,
            None => None,
        }
    }
    /// How long the overall connection establishment should take before a timeout error is returned.
    fn total_connection_timeout(&self) -> Option<Duration> {
        match self.get_peer_options() {
            Some(opt) => opt.total_connection_timeout,
            None => None,
        }
    }
    /// If the connection can be reused, how long the connection should wait to be reused before it
    /// shuts down.
    fn idle_timeout(&self) -> Option<Duration> {
        self.get_peer_options().and_then(|o| o.idle_timeout)
    }

    /// Get the ALPN preference.
    fn get_alpn(&self) -> Option<&ALPN> {
        self.get_peer_options().map(|opt| &opt.alpn)
    }

    /// Get the CA cert to use to validate the server cert.
    ///
    /// If not set, the default CAs will be used.
    fn get_ca(&self) -> Option<&Arc<CaType>> {
        match self.get_peer_options() {
            Some(opt) => opt.ca.as_ref(),
            None => None,
        }
    }

    /// Get the client cert and key for mutual TLS if any
    fn get_client_cert_key(&self) -> Option<&Arc<CertKey>> {
        None
    }

    /// Get the PSK (pre-shared key) to use to validate the connection
    ///
    /// If not set, PSK validation will not be used
    #[cfg(feature = "s2n")]
    fn get_psk(&self) -> Option<&Arc<PskType>> {
        match self.get_peer_options() {
            Some(opt) => opt.psk.as_ref(),
            None => None,
        }
    }

    /// Get the Security Policy to use for this connection (S2N only)
    ///
    /// If not set, the default policy "default_tls13" will be used
    /// https://aws.github.io/s2n-tls/usage-guide/ch06-security-policies.html
    #[cfg(feature = "s2n")]
    fn get_s2n_security_policy(&self) -> Option<&S2NPolicy> {
        match self.get_peer_options() {
            Some(opt) => opt.s2n_security_policy.as_ref(),
            None => None,
        }
    }

    /// S2N-TLS will delay a response up to the max blinding delay (default 30)
    /// seconds whenever an error triggered by a peer occurs to mitigate against
    /// timing side channels.
    #[cfg(feature = "s2n")]
    fn get_max_blinding_delay(&self) -> Option<u32> {
        match self.get_peer_options() {
            Some(opt) => opt.max_blinding_delay,
            None => None,
        }
    }

    /// The TCP keepalive setting that should be applied to this connection
    fn tcp_keepalive(&self) -> Option<&TcpKeepalive> {
        self.get_peer_options()
            .and_then(|o| o.tcp_keepalive.as_ref())
    }

    /// The interval H2 pings to send to the server if any
    fn h2_ping_interval(&self) -> Option<Duration> {
        self.get_peer_options().and_then(|o| o.h2_ping_interval)
    }

    /// The size of the TCP receive buffer should be limited to. See SO_RCVBUF for more details.
    fn tcp_recv_buf(&self) -> Option<usize> {
        self.get_peer_options().and_then(|o| o.tcp_recv_buf)
    }

    /// The DSCP value that should be applied to the send side of this connection.
    /// See the [RFC](https://datatracker.ietf.org/doc/html/rfc2474) for more details.
    fn dscp(&self) -> Option<u8> {
        self.get_peer_options().and_then(|o| o.dscp)
    }

    /// Whether to enable TCP fast open.
    fn tcp_fast_open(&self) -> bool {
        self.get_peer_options()
            .map(|o| o.tcp_fast_open)
            .unwrap_or_default()
    }

    #[cfg(unix)]
    fn matches_fd<V: AsRawFd>(&self, fd: V) -> bool {
        self.address().check_fd_match(fd)
    }

    #[cfg(windows)]
    fn matches_sock<V: AsRawSocket>(&self, sock: V) -> bool {
        use crate::protocols::ConnSockReusable;
        self.address().check_sock_match(sock)
    }

    fn get_tracer(&self) -> Option<Tracer> {
        None
    }

    /// Returns a hook that should be run before an upstream TCP connection is connected.
    ///
    /// This hook can be used to set additional socket options.
    fn upstream_tcp_sock_tweak_hook(
        &self,
    ) -> Option<&Arc<dyn Fn(&TcpSocket) -> Result<()> + Send + Sync + 'static>> {
        self.get_peer_options()?
            .upstream_tcp_sock_tweak_hook
            .as_ref()
    }

    /// Returns a [`ProxyDigestUserDataHook`] that may generate user data for
    /// [`crate::protocols::raw_connect::ProxyDigest`] when establishing a new proxy connection.
    fn proxy_digest_user_data_hook(&self) -> Option<&ProxyDigestUserDataHook> {
        self.get_peer_options()?
            .proxy_digest_user_data_hook
            .as_ref()
    }

    /// Returns a hook that should be run on TLS handshake completion.
    ///
    /// Any value returned from the returned hook (other than `None`) will be stored in the
    /// `extension` field of `SslDigest`. This allows you to attach custom application-specific
    /// data to the TLS connection, which will be accessible from the HTTP layer via the
    /// `SslDigest` attached to the session digest.
    ///
    /// Currently only enabled for openssl variants with meaningful `TlsRef`s.
    #[cfg(feature = "openssl_derived")]
    fn upstream_tls_handshake_complete_hook(&self) -> Option<&HandshakeCompleteHook> {
        self.get_peer_options()?
            .upstream_tls_handshake_complete_hook
            .as_ref()
    }
}

/// A simple TCP or TLS peer without many complicated settings.
#[derive(Debug, Clone)]
pub struct BasicPeer {
    pub _address: SocketAddr,
    pub sni: String,
    pub options: PeerOptions,
}

impl BasicPeer {
    /// Create a new [`BasicPeer`].
    pub fn new(address: &str) -> Self {
        let addr = SocketAddr::Inet(address.parse().unwrap()); // TODO: check error
        Self::new_from_sockaddr(addr)
    }

    /// Create a new [`BasicPeer`] with the given path to a Unix domain socket.
    #[cfg(unix)]
    pub fn new_uds<P: AsRef<Path>>(path: P) -> Result<Self> {
        let addr = SocketAddr::Unix(
            UnixSocketAddr::from_pathname(path.as_ref())
                .or_err(InternalError, "while creating BasicPeer")?,
        );
        Ok(Self::new_from_sockaddr(addr))
    }

    fn new_from_sockaddr(sockaddr: SocketAddr) -> Self {
        BasicPeer {
            _address: sockaddr,
            sni: "".to_string(), // TODO: add support for SNI
            options: PeerOptions::new(),
        }
    }
}

impl Display for BasicPeer {
    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
        write!(f, "{:?}", self)
    }
}

impl Peer for BasicPeer {
    fn address(&self) -> &SocketAddr {
        &self._address
    }

    fn tls(&self) -> bool {
        !self.sni.is_empty()
    }

    fn bind_to(&self) -> Option<&BindTo> {
        None
    }

    fn sni(&self) -> &str {
        &self.sni
    }

    // TODO: change connection pool to accept u64 instead of String
    fn reuse_hash(&self) -> u64 {
        let mut hasher = AHasher::default();
        self._address.hash(&mut hasher);
        hasher.finish()
    }

    fn get_peer_options(&self) -> Option<&PeerOptions> {
        Some(&self.options)
    }
}

/// Define whether to connect via http or https
#[derive(Hash, Clone, Debug, PartialEq)]
pub enum Scheme {
    HTTP,
    HTTPS,
}

impl Display for Scheme {
    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
        match self {
            Scheme::HTTP => write!(f, "HTTP"),
            Scheme::HTTPS => write!(f, "HTTPS"),
        }
    }
}

impl Scheme {
    pub fn from_tls_bool(tls: bool) -> Self {
        if tls {
            Self::HTTPS
        } else {
            Self::HTTP
        }
    }
}

/// Policy for forwarding request headers to HTTP upstreams.
///
/// This policy applies to automatically forwarded downstream request headers. Application code
/// may deliberately alter the resulting request in its upstream request filter.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct HttpUpstreamRequestPolicy {
    /// Strip standard hop-by-hop request fields inherited from the downstream request.
    ///
    /// Standard hop-by-hop framing fields are removed. If a non-empty request body is unframed
    /// after application upstream request filtering, Pingora sends it chunked to an HTTP/1
    /// upstream.
    pub strip_hop_by_hop: bool,
    /// Strip extension fields identified by tokens in the downstream `Connection` header field.
    ///
    /// Requests nominating `Host`, forwarding-origin fields, or pseudo-header-shaped fields are
    /// rejected rather than forwarded with protected metadata removed when this behavior is
    /// enabled.
    pub strip_connection_nominated: bool,
    /// Reject `Connection` nominations that are not a valid HTTP `token` (RFC 9110 §5.6.2), such
    /// as `Connection: "X-Forwarded-For"`, instead of tolerating them as distinct literal fields.
    ///
    /// Defaults to `true`. Exact nominations of protected fields and pseudo-headers are rejected
    /// either way. Only has an effect when
    /// [`strip_connection_nominated`](Self::strip_connection_nominated) is enabled.
    pub reject_malformed_connection_nominations: bool,
    /// Controls forwarding of HTTP/1 protocol upgrade request fields.
    pub h1_upgrade: H1UpgradePolicy,
}

impl HttpUpstreamRequestPolicy {
    /// Use standards-oriented forwarding with normalized WebSocket upgrade support.
    pub fn standard() -> Self {
        Self::default()
    }

    /// Preserve the previous HTTP/1 upstream request-header passthrough behavior.
    ///
    /// This mode is RFC-non-compliant and is provided only for legacy compatibility. Use it at
    /// your own risk: the application's upstream request filter is solely responsible for
    /// ensuring valid hop-by-hop header handling.
    pub fn preserve() -> Self {
        Self {
            strip_hop_by_hop: false,
            strip_connection_nominated: false,
            reject_malformed_connection_nominations: false,
            h1_upgrade: H1UpgradePolicy::Preserve,
        }
    }

    /// Strip hop-by-hop fields and do not forward any HTTP/1 upgrade handshake.
    pub fn deny_upgrades() -> Self {
        Self {
            h1_upgrade: H1UpgradePolicy::Deny,
            ..Self::default()
        }
    }
}

impl Default for HttpUpstreamRequestPolicy {
    fn default() -> Self {
        Self {
            strip_hop_by_hop: true,
            strip_connection_nominated: true,
            reject_malformed_connection_nominations: true,
            h1_upgrade: H1UpgradePolicy::WebSocketOnly,
        }
    }
}

/// Policy for forwarding HTTP/1 protocol upgrade request fields.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum H1UpgradePolicy {
    /// Forward normalized upgrade fields only for a valid WebSocket upgrade request.
    WebSocketOnly,
    /// Preserve complete request metadata for HTTP/1 requests containing an upgrade field.
    ///
    /// This is RFC-non-compliant and is provided only for legacy compatibility. Use it at your
    /// own risk: the application's upstream request filter is solely responsible for ensuring
    /// valid hop-by-hop metadata for upgraded requests.
    ///
    /// When this is selected, automatic hop-by-hop normalization is skipped for upgrade requests
    /// because protocol-specific handshakes can depend on any connection-nominated field.
    /// Protected nomination validation applies only when
    /// [`HttpUpstreamRequestPolicy::strip_connection_nominated`] is enabled.
    Preserve,
    /// Do not forward HTTP/1 upgrade request fields.
    Deny,
}

/// The preferences to connect to a remote server
///
/// See [`Peer`] for the meaning of the fields
#[non_exhaustive]
#[derive(Clone, Derivative)]
#[derivative(Debug)]
pub struct PeerOptions {
    pub bind_to: Option<BindTo>,
    pub connection_timeout: Option<Duration>,
    pub total_connection_timeout: Option<Duration>,
    pub read_timeout: Option<Duration>,
    pub idle_timeout: Option<Duration>,
    pub write_timeout: Option<Duration>,
    pub verify_cert: bool,
    pub verify_hostname: bool,
    #[cfg(feature = "s2n")]
    pub use_system_certs: bool,
    /* accept the cert if it's CN matches the SNI or this name */
    pub alternative_cn: Option<String>,
    pub alpn: ALPN,
    pub ca: Option<Arc<CaType>>,
    pub tcp_keepalive: Option<TcpKeepalive>,
    pub tcp_recv_buf: Option<usize>,
    pub dscp: Option<u8>,
    pub h2_ping_interval: Option<Duration>,
    #[cfg(feature = "s2n")]
    pub psk: Option<Arc<PskType>>,
    #[cfg(feature = "s2n")]
    pub s2n_security_policy: Option<S2NPolicy>,
    #[cfg(feature = "s2n")]
    pub max_blinding_delay: Option<u32>,
    /// How many concurrent h2 streams are allowed in the same connection.
    pub max_h2_streams: usize,
    /// Initial per-stream H2 receive window size in bytes.
    /// If `None`, the default of 8MB is used.
    pub h2_stream_window_size: Option<u32>,
    /// Initial connection-level H2 receive window size in bytes.
    /// If `None`, the default of 8MB is used.
    pub h2_connection_window_size: Option<u32>,
    /// Allow a single invalid Content-Length in HTTP/1 responses (non-RFC compliant).
    ///
    /// When enabled, a response carrying a single, otherwise-unparseable
    /// Content-Length value is treated as a close-delimited response instead of
    /// being rejected. Conflicting or duplicate Content-Length values (multiple
    /// header lines, or a comma-separated list with differing values) are an
    /// unrecoverable framing error and are always rejected, even when this is
    /// enabled.
    ///
    /// **Note:** This field is unstable and may be removed or changed in future versions.
    /// It exists primarily for compatibility with legacy servers that send malformed headers.
    pub allow_h1_response_invalid_content_length: bool,
    /// Controls automatically forwarded request headers sent to HTTP upstreams.
    pub http_upstream_request_policy: HttpUpstreamRequestPolicy,
    pub extra_proxy_headers: BTreeMap<String, Vec<u8>>,
    /// The list of curves the tls connection should advertise
    /// if `None`, the default curves will be used
    pub curves: Option<Cow<'static, str>>,
    /// see ssl_use_second_key_share
    pub second_keyshare: bool,
    /// whether to enable TCP fast open
    pub tcp_fast_open: bool,
    /// use Arc because Clone is required but not allowed in trait object
    pub tracer: Option<Tracer>,
    /// A custom L4 connector to use to establish new L4 connections
    pub custom_l4: Option<Arc<dyn L4Connect + Send + Sync>>,
    #[derivative(Debug = "ignore")]
    pub upstream_tcp_sock_tweak_hook:
        Option<Arc<dyn Fn(&TcpSocket) -> Result<()> + Send + Sync + 'static>>,
    #[derivative(Debug = "ignore")]
    pub proxy_digest_user_data_hook: Option<ProxyDigestUserDataHook>,
    /// Hook that allows returning an optional `SslDigestExtension`.
    /// Any returned value will be saved into the `SslDigest`.
    ///
    /// Currently only enabled for openssl variants with meaningful `TlsRef`s.
    #[cfg(feature = "openssl_derived")]
    #[derivative(Debug = "ignore")]
    pub upstream_tls_handshake_complete_hook: Option<HandshakeCompleteHook>,
}

impl PeerOptions {
    /// Create a new [`PeerOptions`]
    pub fn new() -> Self {
        PeerOptions {
            bind_to: None,
            connection_timeout: None,
            total_connection_timeout: None,
            read_timeout: None,
            idle_timeout: None,
            write_timeout: None,
            verify_cert: true,
            verify_hostname: true,
            #[cfg(feature = "s2n")]
            use_system_certs: true,
            alternative_cn: None,
            alpn: ALPN::H1,
            ca: None,
            tcp_keepalive: None,
            tcp_recv_buf: None,
            dscp: None,
            h2_ping_interval: None,
            #[cfg(feature = "s2n")]
            psk: None,
            #[cfg(feature = "s2n")]
            s2n_security_policy: None,
            #[cfg(feature = "s2n")]
            max_blinding_delay: None,
            max_h2_streams: 1,
            h2_stream_window_size: None,
            h2_connection_window_size: None,
            allow_h1_response_invalid_content_length: false,
            http_upstream_request_policy: HttpUpstreamRequestPolicy::default(),
            extra_proxy_headers: BTreeMap::new(),
            curves: None,
            second_keyshare: true, // default true and noop when not using PQ curves
            tcp_fast_open: false,
            tracer: None,
            custom_l4: None,
            upstream_tcp_sock_tweak_hook: None,
            proxy_digest_user_data_hook: None,
            #[cfg(feature = "openssl_derived")]
            upstream_tls_handshake_complete_hook: None,
        }
    }

    /// Set the ALPN according to the `max` and `min` constrains.
    pub fn set_http_version(&mut self, max: u8, min: u8) {
        self.alpn = ALPN::new(max, min);
    }
}

impl Display for PeerOptions {
    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
        if let Some(b) = self.bind_to.as_ref() {
            write!(f, "bind_to: {:?},", b)?;
        }
        if let Some(t) = self.connection_timeout {
            write!(f, "conn_timeout: {:?},", t)?;
        }
        if let Some(t) = self.total_connection_timeout {
            write!(f, "total_conn_timeout: {:?},", t)?;
        }
        if self.verify_cert {
            write!(f, "verify_cert: true,")?;
        }
        if self.verify_hostname {
            write!(f, "verify_hostname: true,")?;
        }
        #[cfg(feature = "s2n")]
        if self.use_system_certs {
            write!(f, "use_system_certs: true,")?;
        }
        if let Some(cn) = &self.alternative_cn {
            write!(f, "alt_cn: {},", cn)?;
        }
        write!(f, "alpn: {},", self.alpn)?;
        if let Some(cas) = &self.ca {
            for ca in cas.iter() {
                write!(
                    f,
                    "CA: {}, expire: {},",
                    get_organization_unit(ca).unwrap_or_default(),
                    ca.not_after()
                )?;
            }
        }
        #[cfg(feature = "s2n")]
        if let Some(policy) = &self.s2n_security_policy {
            write!(f, "s2n_security_policy: {:?}, ", policy)?;
        }
        #[cfg(feature = "s2n")]
        if let Some(psk_config) = &self.psk {
            for psk in &psk_config.keys {
                write!(
                    f,
                    "psk_identity: {}",
                    String::from_utf8_lossy(psk.identity.as_slice())
                )?;
            }
        }
        if let Some(tcp_keepalive) = &self.tcp_keepalive {
            write!(f, "tcp_keepalive: {},", tcp_keepalive)?;
        }
        if let Some(h2_ping_interval) = self.h2_ping_interval {
            write!(f, "h2_ping_interval: {:?},", h2_ping_interval)?;
        }
        Ok(())
    }
}

/// A peer representing the remote HTTP server to connect to
#[derive(Debug, Clone)]
pub struct HttpPeer {
    pub _address: SocketAddr,
    pub scheme: Scheme,
    pub sni: String,
    pub proxy: Option<Proxy>,
    pub client_cert_key: Option<Arc<CertKey>>,
    /// a custom field to isolate connection reuse. Requests with different group keys
    /// cannot share connections with each other.
    pub group_key: u64,
    pub options: PeerOptions,
}

impl HttpPeer {
    // These methods are pretty ad-hoc
    pub fn is_tls(&self) -> bool {
        match self.scheme {
            Scheme::HTTP => false,
            Scheme::HTTPS => true,
        }
    }

    fn new_from_sockaddr(address: SocketAddr, tls: bool, sni: String) -> Self {
        HttpPeer {
            _address: address,
            scheme: Scheme::from_tls_bool(tls),
            sni,
            proxy: None,
            client_cert_key: None,
            group_key: 0,
            options: PeerOptions::new(),
        }
    }

    /// Create a new [`HttpPeer`] with the given socket address and TLS settings.
    pub fn new<A: ToInetSocketAddrs>(address: A, tls: bool, sni: String) -> Self {
        let mut addrs_iter = address.to_socket_addrs().unwrap(); //TODO: handle error
        let addr = addrs_iter.next().unwrap();
        Self::new_from_sockaddr(SocketAddr::Inet(addr), tls, sni)
    }

    /// Create a new [`HttpPeer`] with the given path to Unix domain socket and TLS settings.
    #[cfg(unix)]
    pub fn new_uds(path: &str, tls: bool, sni: String) -> Result<Self> {
        let addr = SocketAddr::Unix(
            UnixSocketAddr::from_pathname(Path::new(path)).or_err(SocketError, "invalid path")?,
        );
        Ok(Self::new_from_sockaddr(addr, tls, sni))
    }

    /// Create a new [`HttpPeer`] that uses a proxy to connect to the upstream IP and port
    /// combination.
    pub fn new_proxy(
        next_hop: &str,
        ip_addr: IpAddr,
        port: u16,
        tls: bool,
        sni: &str,
        headers: BTreeMap<String, Vec<u8>>,
    ) -> Self {
        HttpPeer {
            _address: SocketAddr::Inet(InetSocketAddr::new(ip_addr, port)),
            scheme: Scheme::from_tls_bool(tls),
            sni: sni.to_string(),
            proxy: Some(Proxy {
                next_hop: PathBuf::from(next_hop).into(),
                host: ip_addr.to_string(),
                port,
                headers,
            }),
            client_cert_key: None,
            group_key: 0,
            options: PeerOptions::new(),
        }
    }

    /// Create a new [`HttpPeer`] with client certificate and key for mutual TLS.
    pub fn new_mtls<A: ToInetSocketAddrs>(
        address: A,
        sni: String,
        client_cert_key: Arc<CertKey>,
    ) -> Self {
        let mut peer = Self::new(address, true, sni);
        peer.client_cert_key = Some(client_cert_key);
        peer
    }

    fn peer_hash(&self) -> u64 {
        let mut hasher = AHasher::default();
        self.hash(&mut hasher);
        hasher.finish()
    }
}

impl Hash for HttpPeer {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self._address.hash(state);
        self.scheme.hash(state);
        self.proxy.hash(state);
        self.sni.hash(state);
        // client cert serial
        self.client_cert_key.hash(state);
        // origin server cert verification
        self.verify_cert().hash(state);
        self.verify_hostname().hash(state);
        self.alternative_cn().hash(state);
        #[cfg(feature = "s2n")]
        self.get_psk().hash(state);
        self.group_key.hash(state);
        // max h2 stream settings
        self.options.max_h2_streams.hash(state);
        // h2_stream_window_size and h2_connection_window_size are intentionally excluded
        // from the reuse hash for now. These are per-connection settings applied at handshake
        // time and may be revisited alongside other h2 settings that could be dynamically
        // adjusted over the lifetime of a connection.
        self.options.curves.hash(state);
        self.options.second_keyshare.hash(state);
    }
}

impl Display for HttpPeer {
    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
        write!(f, "addr: {}, scheme: {}", self._address, self.scheme)?;
        if !self.sni.is_empty() {
            write!(f, ", sni: {}", self.sni)?;
        }
        if let Some(p) = self.proxy.as_ref() {
            write!(f, ", proxy: {p}")?;
        }
        if let Some(cert) = &self.client_cert_key {
            write!(f, ", client cert: {}", cert)?;
        }
        Ok(())
    }
}

impl Peer for HttpPeer {
    fn address(&self) -> &SocketAddr {
        &self._address
    }

    fn tls(&self) -> bool {
        self.is_tls()
    }

    fn sni(&self) -> &str {
        &self.sni
    }

    // TODO: change connection pool to accept u64 instead of String
    fn reuse_hash(&self) -> u64 {
        self.peer_hash()
    }

    fn get_peer_options(&self) -> Option<&PeerOptions> {
        Some(&self.options)
    }

    fn get_mut_peer_options(&mut self) -> Option<&mut PeerOptions> {
        Some(&mut self.options)
    }

    fn get_proxy(&self) -> Option<&Proxy> {
        self.proxy.as_ref()
    }

    #[cfg(unix)]
    fn matches_fd<V: AsRawFd>(&self, fd: V) -> bool {
        if let Some(proxy) = self.get_proxy() {
            proxy.next_hop.check_fd_match(fd)
        } else {
            self.address().check_fd_match(fd)
        }
    }

    #[cfg(windows)]
    fn matches_sock<V: AsRawSocket>(&self, sock: V) -> bool {
        use crate::protocols::ConnSockReusable;

        if let Some(proxy) = self.get_proxy() {
            panic!("windows do not support peers with proxy")
        } else {
            self.address().check_sock_match(sock)
        }
    }

    fn get_client_cert_key(&self) -> Option<&Arc<CertKey>> {
        self.client_cert_key.as_ref()
    }

    fn get_tracer(&self) -> Option<Tracer> {
        self.options.tracer.clone()
    }
}

/// The proxy settings to connect to the remote server, CONNECT only for now
#[derive(Debug, Hash, Clone)]
pub struct Proxy {
    pub next_hop: Box<Path>, // for now this will be the path to the UDS
    pub host: String,        // the proxied host. Could be either IP addr or hostname.
    pub port: u16,           // the port to proxy to
    pub headers: BTreeMap<String, Vec<u8>>, // the additional headers to add to CONNECT
}

impl Display for Proxy {
    fn fmt(&self, f: &mut Formatter) -> FmtResult {
        write!(
            f,
            "next_hop: {}, host: {}, port: {}",
            self.next_hop.display(),
            self.host,
            self.port
        )
    }
}

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

    #[test]
    fn default_http_upstream_request_policy_is_standards_oriented() {
        let policy = PeerOptions::new().http_upstream_request_policy;
        assert!(policy.strip_hop_by_hop);
        assert!(policy.strip_connection_nominated);
        assert!(policy.reject_malformed_connection_nominations);
        assert_eq!(policy.h1_upgrade, H1UpgradePolicy::WebSocketOnly);
    }

    #[test]
    fn preserve_http_upstream_request_policy_is_a_legacy_preset() {
        let policy = HttpUpstreamRequestPolicy::preserve();
        assert!(!policy.strip_hop_by_hop);
        assert!(!policy.strip_connection_nominated);
        assert!(!policy.reject_malformed_connection_nominations);
        assert_eq!(policy.h1_upgrade, H1UpgradePolicy::Preserve);
    }
}