spin_factor_outbound_http/

wasi.rs

use std::{
    error::Error,
    future::Future,
    io::IoSlice,
    net::SocketAddr,
    ops::DerefMut,
    pin::Pin,
    sync::{Arc, Mutex},
    task::{self, Context, Poll},
    time::Duration,
};

use bytes::{Buf, Bytes};
use futures::channel::oneshot;
use http::{
    HeaderMap, Uri,
    header::{CONTENT_LENGTH, HOST},
    uri::Scheme,
};
use http_body::{Body, Frame, SizeHint};
use http_body_util::{BodyExt, combinators::UnsyncBoxBody};
use hyper_util::{
    client::legacy::{
        Client,
        connect::{Connected, Connection},
    },
    rt::{TokioExecutor, TokioIo},
};
use opentelemetry_semantic_conventions::attribute as otel_attribute;
use spin_factor_outbound_networking::{
    ComponentTlsClientConfigs, TlsClientConfig,
    config::{allowed_hosts::OutboundAllowedHosts, blocked_networks::BlockedNetworks},
};
use spin_factors::RuntimeFactorsInstanceState;
use tokio::{
    io::{AsyncRead, AsyncWrite, ReadBuf},
    net::TcpStream,
    time::timeout,
};
use tokio_rustls::client::TlsStream;
use tower_service::Service;
use tracing::{Instrument, Span, field::Empty, instrument};
use wasmtime::component::HasData;
use wasmtime_wasi::TrappableError;
use wasmtime_wasi_http::{
    p2::{
        self, HttpError, WasiHttpCtxView,
        bindings::http::types::{self as p2_types, ErrorCode},
        body::HyperOutgoingBody,
        types::{HostFutureIncomingResponse, IncomingResponse, OutgoingRequestConfig},
    },
    p3::{self, bindings::http::types as p3_types},
};

use spin_factor_outbound_networking::{ConnectionPermit, ConnectionSemaphore};

use crate::{
    InstanceHttpHooks, OutboundHttpFactor, SelfRequestOrigin,
    intercept::{InterceptOutcome, OutboundHttpInterceptor},
    wasi_2023_10_18, wasi_2023_11_10,
};

use tracing_opentelemetry::OpenTelemetrySpanExt as _;

const DEFAULT_TIMEOUT: Duration = Duration::from_secs(600);

pub struct MutexBody<T>(Mutex<T>);

impl<T> MutexBody<T> {
    pub fn new(body: T) -> Self {
        Self(Mutex::new(body))
    }
}

impl<T: Body + Unpin> Body for MutexBody<T> {
    type Data = T::Data;
    type Error = T::Error;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        Pin::new(self.0.lock().unwrap().deref_mut()).poll_frame(cx)
    }

    fn is_end_stream(&self) -> bool {
        self.0.lock().unwrap().is_end_stream()
    }

    fn size_hint(&self) -> SizeHint {
        self.0.lock().unwrap().size_hint()
    }
}

/// Body which, when dropped, will notify the receiver corresponding to the
/// specified sender.
pub struct NotifyOnDropBody<B> {
    body: B,
    _tx: oneshot::Sender<()>,
}

impl<B> NotifyOnDropBody<B> {
    pub fn new(body: B, tx: oneshot::Sender<()>) -> Self {
        Self { body, _tx: tx }
    }
}

impl<B: Body + Unpin> Body for NotifyOnDropBody<B> {
    type Data = B::Data;
    type Error = B::Error;

    fn poll_frame(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        Pin::new(&mut self.body).poll_frame(cx)
    }

    fn is_end_stream(&self) -> bool {
        self.body.is_end_stream()
    }

    fn size_hint(&self) -> SizeHint {
        self.body.size_hint()
    }
}

pub(crate) struct HasHttp;

impl HasData for HasHttp {
    type Data<'a> = WasiHttpCtxView<'a>;
}

impl p3::WasiHttpHooks for InstanceHttpHooks {
    #[instrument(
        name = "spin_outbound_http.send_request",
        skip_all,
        fields(
            otel.kind = "client",
            {otel_attribute::URL_FULL} = Empty,
            {otel_attribute::HTTP_REQUEST_METHOD} = %request.method(),
            otel.name = %request.method(),
            // Incubating convention; not yet a stable `opentelemetry_semantic_conventions` constant.
            http.response.body.size = Empty,
            {otel_attribute::HTTP_RESPONSE_STATUS_CODE} = Empty,
            {otel_attribute::SERVER_ADDRESS} = Empty,
            {otel_attribute::SERVER_PORT} = Empty,
        )
    )]
    #[allow(clippy::type_complexity)]
    fn send_request(
        &mut self,
        request: http::Request<UnsyncBoxBody<Bytes, p3_types::ErrorCode>>,
        options: Option<p3::RequestOptions>,
        fut: Box<dyn Future<Output = Result<(), p3_types::ErrorCode>> + Send>,
    ) -> Box<
        dyn Future<
                Output = Result<
                    (
                        http::Response<UnsyncBoxBody<Bytes, p3_types::ErrorCode>>,
                        Box<dyn Future<Output = Result<(), p3_types::ErrorCode>> + Send>,
                    ),
                    TrappableError<p3_types::ErrorCode>,
                >,
            > + Send,
    > {
        self.otel.reparent_tracing_span();

        // If the caller (i.e. the guest) has trouble consuming the response
        // (e.g. encountering a network error while forwarding it on to some
        // other place), it can report that error to us via `fut`.  However,
        // there's nothing we'll be able to do with it here, so we ignore it.
        // Presumably the guest will also drop the body stream and trailers
        // future if it encounters such an error while those things are still
        // arriving, which Hyper will deal with as appropriate (e.g. closing the
        // connection).
        _ = fut;

        let request = request.map(|body| MutexBody::new(body).boxed());

        let request_sender = RequestSender {
            allowed_hosts: self.allowed_hosts.clone(),
            component_tls_configs: self.component_tls_configs.clone(),
            request_interceptor: self.request_interceptor.clone(),
            self_request_origin: self.self_request_origin.clone(),
            blocked_networks: self.blocked_networks.clone(),
            http_clients: self.wasi_http_clients.clone(),
            semaphore: self.semaphore.clone(),
        };
        let config = OutgoingRequestConfig {
            use_tls: request.uri().scheme() == Some(&Scheme::HTTPS),
            connect_timeout: options
                .and_then(|v| v.connect_timeout)
                .unwrap_or(DEFAULT_TIMEOUT),
            first_byte_timeout: options
                .and_then(|v| v.first_byte_timeout)
                .unwrap_or(DEFAULT_TIMEOUT),
            between_bytes_timeout: options
                .and_then(|v| v.between_bytes_timeout)
                .unwrap_or(DEFAULT_TIMEOUT),
        };
        Box::new(
            async {
                match request_sender
                    .send(
                        request.map(|body| body.map_err(p3_to_p2_error_code).boxed_unsync()),
                        config,
                    )
                    .await
                {
                    Ok(IncomingResponse {
                        resp,
                        between_bytes_timeout,
                        ..
                    }) => Ok((
                        resp.map(|body| {
                            BetweenBytesTimeoutBody {
                                body: Some(body),
                                sleep: None,
                                timeout: between_bytes_timeout,
                                byte_count: 0,
                                span: Some(Span::current()),
                            }
                            .boxed_unsync()
                        }),
                        Box::new(async {
                            // TODO: Can we plumb connection errors through to here, or
                            // will `hyper_util::client::legacy::Client` pass them all
                            // via the response body?
                            Ok(())
                        }) as Box<dyn Future<Output = _> + Send>,
                    )),
                    Err(http_error) => match http_error.downcast() {
                        Ok(error_code) => {
                            Err(TrappableError::from(p2_to_p3_error_code(error_code)))
                        }
                        Err(trap) => Err(TrappableError::trap(trap)),
                    },
                }
            }
            .in_current_span(),
        )
    }
}

pin_project_lite::pin_project! {
    struct BetweenBytesTimeoutBody<B> {
        // Wrapping `body` in `Option` lets us take it out (dropping the underlying connection) when the timeout fires.
        body: Option<B>,
        #[pin]
        sleep: Option<tokio::time::Sleep>,
        timeout: Duration,
        byte_count: u64,
        span: Option<Span>,
    }
}

impl<B: Body<Error = p2_types::ErrorCode> + Unpin> Body for BetweenBytesTimeoutBody<B> {
    type Data = B::Data;
    type Error = p3_types::ErrorCode;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        let mut me = self.project();

        // Scope the mutable borrow so we can touch `me.body` again below.
        let poll_result = {
            let Some(body) = me.body.as_mut() else {
                // Body already dropped by a previous timeout fire.
                return Poll::Ready(None);
            };
            Pin::new(body).poll_frame(cx)
        };

        let mut record_body_size_once = |body_size: u64| {
            if let Some(span) = me.span.take() {
                // `http.response.body.size` is incubating (behind semconv_experimental)
                // in opentelemetry-semantic-conventions 0.28. Leave as literal to avoid
                // enabling the experimental feature.
                span.record("http.response.body.size", body_size);
            }
        };
        match poll_result {
            Poll::Ready(value) => {
                me.sleep.as_mut().set(None);

                match &value {
                    Some(Ok(frame)) => {
                        if let Some(data) = frame.data_ref() {
                            *me.byte_count += data.remaining() as u64;
                        }
                        if me.body.as_ref().is_some_and(|b| b.is_end_stream()) {
                            record_body_size_once(*me.byte_count);
                        }
                    }
                    None => {
                        record_body_size_once(*me.byte_count);
                    }
                    Some(Err(e)) => {
                        tracing::warn!("error reading response body: {e:?}");
                    }
                }

                Poll::Ready(value.map(|v| v.map_err(p2_to_p3_error_code)))
            }
            Poll::Pending => {
                if me.sleep.is_none() {
                    me.sleep.as_mut().set(Some(tokio::time::sleep(*me.timeout)));
                }
                task::ready!(me.sleep.as_pin_mut().unwrap().poll(cx));

                // Drop the inner body immediately to free resources (like sockets)
                // rather than waiting for the guest to release the resource.
                *me.body = None;
                record_body_size_once(*me.byte_count);

                Poll::Ready(Some(Err(p3_types::ErrorCode::ConnectionReadTimeout)))
            }
        }
    }

    fn is_end_stream(&self) -> bool {
        self.body.as_ref().is_none_or(|b| b.is_end_stream())
    }

    fn size_hint(&self) -> SizeHint {
        self.body
            .as_ref()
            .map(|b| b.size_hint())
            .unwrap_or_default()
    }
}

pub(crate) fn add_to_linker<C>(ctx: &mut C) -> anyhow::Result<()>
where
    C: spin_factors::InitContext<OutboundHttpFactor>,
{
    let linker = ctx.linker();

    fn get_http<C>(store: &mut C::StoreData) -> WasiHttpCtxView<'_>
    where
        C: spin_factors::InitContext<OutboundHttpFactor>,
    {
        let (state, table) = C::get_data_with_table(store);
        let ctx = &mut state.wasi_http_ctx;
        WasiHttpCtxView {
            ctx,
            table,
            hooks: &mut state.hooks,
        }
    }

    let get_http = get_http::<C> as fn(&mut C::StoreData) -> WasiHttpCtxView<'_>;
    wasmtime_wasi_http::p2::bindings::http::outgoing_handler::add_to_linker::<_, HasHttp>(
        linker, get_http,
    )?;
    wasmtime_wasi_http::p2::bindings::http::types::add_to_linker::<_, HasHttp>(
        linker,
        &Default::default(),
        get_http,
    )?;

    fn get_http_p3<C>(store: &mut C::StoreData) -> p3::WasiHttpCtxView<'_>
    where
        C: spin_factors::InitContext<OutboundHttpFactor>,
    {
        let (state, table) = C::get_data_with_table(store);
        let ctx = &mut state.wasi_http_ctx;
        p3::WasiHttpCtxView {
            ctx,
            table,
            hooks: &mut state.hooks,
        }
    }

    let get_http_p3 = get_http_p3::<C> as fn(&mut C::StoreData) -> p3::WasiHttpCtxView<'_>;
    p3::bindings::http::client::add_to_linker::<_, p3::WasiHttp>(linker, get_http_p3)?;
    p3::bindings::http::types::add_to_linker::<_, p3::WasiHttp>(linker, get_http_p3)?;

    wasi_2023_10_18::add_to_linker(linker, get_http)?;
    wasi_2023_11_10::add_to_linker(linker, get_http)?;

    Ok(())
}

impl OutboundHttpFactor {
    pub fn get_wasi_http_impl(
        runtime_instance_state: &mut impl RuntimeFactorsInstanceState,
    ) -> Option<WasiHttpCtxView<'_>> {
        let (state, table) = runtime_instance_state.get_with_table::<OutboundHttpFactor>()?;
        let ctx = &mut state.wasi_http_ctx;
        Some(WasiHttpCtxView {
            ctx,
            table,
            hooks: &mut state.hooks,
        })
    }

    pub fn get_wasi_p3_http_impl(
        runtime_instance_state: &mut impl RuntimeFactorsInstanceState,
    ) -> Option<p3::WasiHttpCtxView<'_>> {
        let (state, table) = runtime_instance_state.get_with_table::<OutboundHttpFactor>()?;
        let ctx = &mut state.wasi_http_ctx;
        Some(p3::WasiHttpCtxView {
            ctx,
            table,
            hooks: &mut state.hooks,
        })
    }
}

type OutgoingRequest = http::Request<HyperOutgoingBody>;

impl p2::WasiHttpHooks for InstanceHttpHooks {
    #[instrument(
        name = "spin_outbound_http.send_request",
        skip_all,
        fields(
            otel.kind = "client",
            {otel_attribute::URL_FULL} = Empty,
            {otel_attribute::HTTP_REQUEST_METHOD} = %request.method(),
            otel.name = %request.method(),
            {otel_attribute::HTTP_RESPONSE_STATUS_CODE} = Empty,
            {otel_attribute::SERVER_ADDRESS} = Empty,
            {otel_attribute::SERVER_PORT} = Empty,
        )
    )]
    fn send_request(
        &mut self,
        request: OutgoingRequest,
        config: OutgoingRequestConfig,
    ) -> Result<wasmtime_wasi_http::p2::types::HostFutureIncomingResponse, HttpError> {
        self.otel.reparent_tracing_span();

        let request_sender = RequestSender {
            allowed_hosts: self.allowed_hosts.clone(),
            component_tls_configs: self.component_tls_configs.clone(),
            request_interceptor: self.request_interceptor.clone(),
            self_request_origin: self.self_request_origin.clone(),
            blocked_networks: self.blocked_networks.clone(),
            http_clients: self.wasi_http_clients.clone(),
            semaphore: self.semaphore.clone(),
        };
        Ok(HostFutureIncomingResponse::Pending(
            wasmtime_wasi::runtime::spawn(
                async {
                    match request_sender.send(request, config).await {
                        Ok(resp) => Ok(Ok(resp)),
                        Err(http_error) => match http_error.downcast() {
                            Ok(error_code) => Ok(Err(error_code)),
                            Err(trap) => Err(trap),
                        },
                    }
                }
                .in_current_span(),
            ),
        ))
    }
}

struct RequestSender {
    allowed_hosts: OutboundAllowedHosts,
    blocked_networks: BlockedNetworks,
    component_tls_configs: ComponentTlsClientConfigs,
    self_request_origin: Option<SelfRequestOrigin>,
    request_interceptor: Option<Arc<dyn OutboundHttpInterceptor>>,
    http_clients: HttpClients,
    semaphore: ConnectionSemaphore,
}

impl RequestSender {
    async fn send(
        self,
        mut request: OutgoingRequest,
        mut config: OutgoingRequestConfig,
    ) -> Result<IncomingResponse, HttpError> {
        self.prepare_request(&mut request, &mut config).await?;

        // If the current span has opentelemetry trace context, inject it into the request
        spin_telemetry::inject_trace_context(&mut request);

        // Run any configured request interceptor
        let mut override_connect_addr = None;
        if let Some(interceptor) = &self.request_interceptor {
            let intercept_request = std::mem::take(&mut request).into();
            match interceptor.intercept(intercept_request).await? {
                InterceptOutcome::Continue(mut req) => {
                    override_connect_addr = req.override_connect_addr.take();
                    request = req.into_hyper_request();
                }
                InterceptOutcome::Complete(resp) => {
                    let resp = IncomingResponse {
                        resp,
                        worker: None,
                        between_bytes_timeout: config.between_bytes_timeout,
                    };
                    return Ok(resp);
                }
            }
        }

        // Backfill span fields after potentially updating the URL in the interceptor
        let span = tracing::Span::current();
        if let Some(addr) = override_connect_addr {
            span.record(otel_attribute::SERVER_ADDRESS, addr.ip().to_string());
            span.record(otel_attribute::SERVER_PORT, addr.port());
        } else if let Some(authority) = request.uri().authority() {
            span.record(otel_attribute::SERVER_ADDRESS, authority.host());
            if let Some(port) = authority.port_u16() {
                span.record(otel_attribute::SERVER_PORT, port);
            }
        }

        record_content_length_header(
            &span,
            request.headers(),
            "http.request.header.content-length",
        );

        Ok(self
            .send_request(request, config, override_connect_addr)
            .await?)
    }

    async fn prepare_request(
        &self,
        request: &mut OutgoingRequest,
        config: &mut OutgoingRequestConfig,
    ) -> Result<(), ErrorCode> {
        // wasmtime-wasi-http fills in scheme and authority for relative URLs
        // (e.g. https://:443/<path>), which makes them hard to reason about.
        // Undo that here.
        let uri = request.uri_mut();
        if uri
            .authority()
            .is_some_and(|authority| authority.host().is_empty())
        {
            let mut builder = http::uri::Builder::new();
            if let Some(paq) = uri.path_and_query() {
                builder = builder.path_and_query(paq.clone());
            }
            *uri = builder.build().unwrap();
        }
        tracing::Span::current().record(otel_attribute::URL_FULL, uri.to_string());

        let is_self_request = match request.uri().authority() {
            // Some SDKs require an authority, so we support e.g. http://self.alt/self-request
            Some(authority) => authority.host() == "self.alt",
            // Otherwise self requests have no authority
            None => true,
        };

        // Enforce allowed_outbound_hosts
        let is_allowed = if is_self_request {
            self.allowed_hosts
                .check_relative_url(&["http", "https"])
                .await
                .unwrap_or(false)
        } else {
            self.allowed_hosts
                .check_url(&request.uri().to_string(), "https")
                .await
                .unwrap_or(false)
        };
        if !is_allowed {
            return Err(ErrorCode::HttpRequestDenied);
        }

        if is_self_request {
            // Replace the authority with the "self request origin"
            let Some(origin) = self.self_request_origin.as_ref() else {
                tracing::error!(
                    "Couldn't handle outbound HTTP request to relative URI; no origin set"
                );
                return Err(ErrorCode::HttpRequestUriInvalid);
            };

            config.use_tls = origin.use_tls();

            request.headers_mut().insert(HOST, origin.host_header());

            let path_and_query = request.uri().path_and_query().cloned();
            *request.uri_mut() = origin.clone().into_uri(path_and_query);
        }

        // Some servers (looking at you nginx) don't like a host header even though
        // http/2 allows it: https://github.com/hyperium/hyper/issues/3298.
        //
        // Note that we do this _before_ invoking the request interceptor.  It may
        // decide to add the `host` header back in, regardless of the nginx bug, in
        // which case we'll let it do so without interferring.
        request.headers_mut().remove(HOST);
        Ok(())
    }

    async fn send_request(
        self,
        request: OutgoingRequest,
        config: OutgoingRequestConfig,
        override_connect_addr: Option<SocketAddr>,
    ) -> Result<IncomingResponse, ErrorCode> {
        let OutgoingRequestConfig {
            use_tls,
            connect_timeout,
            first_byte_timeout,
            between_bytes_timeout,
        } = config;

        let tls_client_config = if use_tls {
            let host = request.uri().host().unwrap_or_default();
            Some(self.component_tls_configs.get_client_config(host).clone())
        } else {
            None
        };

        let resp = CONNECT_OPTIONS.scope(
            ConnectOptions {
                blocked_networks: self.blocked_networks,
                connect_timeout,
                tls_client_config,
                override_connect_addr,
                semaphore: self.semaphore,
            },
            async move {
                if use_tls {
                    self.http_clients.https.request(request).await
                } else {
                    // For development purposes, allow configuring plaintext HTTP/2 for a specific host.
                    let h2c_prior_knowledge_host =
                        std::env::var("SPIN_OUTBOUND_H2C_PRIOR_KNOWLEDGE").ok();
                    let use_h2c = h2c_prior_knowledge_host.as_deref()
                        == request.uri().authority().map(|a| a.as_str());

                    if use_h2c {
                        self.http_clients.http2.request(request).await
                    } else {
                        self.http_clients.http1.request(request).await
                    }
                }
            },
        );

        let resp = timeout(first_byte_timeout, resp)
            .await
            .map_err(|_| ErrorCode::ConnectionReadTimeout)?
            .map_err(hyper_legacy_request_error)?
            .map(|body| body.map_err(hyper_request_error).boxed_unsync());

        let span = tracing::Span::current();
        span.record(
            otel_attribute::HTTP_RESPONSE_STATUS_CODE,
            resp.status().as_u16(),
        );

        record_content_length_header(&span, resp.headers(), "http.response.header.content-length");

        Ok(IncomingResponse {
            resp,
            worker: None,
            between_bytes_timeout,
        })
    }
}

type HttpClient = Client<HttpConnector, HyperOutgoingBody>;
type HttpsClient = Client<HttpsConnector, HyperOutgoingBody>;

#[derive(Clone)]
pub(super) struct HttpClients {
    /// Used for non-TLS HTTP/1 connections.
    http1: HttpClient,
    /// Used for non-TLS HTTP/2 connections (e.g. when h2 prior knowledge is available).
    http2: HttpClient,
    /// Used for HTTP-over-TLS connections, using ALPN to negotiate the HTTP version.
    https: HttpsClient,
}

impl HttpClients {
    pub(super) fn new(enable_pooling: bool) -> Self {
        let builder = move || {
            let mut builder = Client::builder(TokioExecutor::new());
            if !enable_pooling {
                builder.pool_max_idle_per_host(0);
            }
            builder
        };
        Self {
            http1: builder().build(HttpConnector),
            http2: builder().http2_only(true).build(HttpConnector),
            https: builder().build(HttpsConnector),
        }
    }
}

tokio::task_local! {
    /// The options used when establishing a new connection.
    ///
    /// We must use task-local variables for these config options when using
    /// `hyper_util::client::legacy::Client::request` because there's no way to plumb
    /// them through as parameters.  Moreover, if there's already a pooled connection
    /// ready, we'll reuse that and ignore these options anyway. After each connection
    /// is established, the options are dropped.
    static CONNECT_OPTIONS: ConnectOptions;
}

#[derive(Clone)]
struct ConnectOptions {
    /// The blocked networks configuration.
    blocked_networks: BlockedNetworks,
    /// Timeout for establishing a TCP connection.
    connect_timeout: Duration,
    /// TLS client configuration to use, if any.
    tls_client_config: Option<TlsClientConfig>,
    /// If set, override the address to connect to instead of using the given `uri`'s authority.
    override_connect_addr: Option<SocketAddr>,
    /// Semaphore to limit concurrent outbound connections.
    semaphore: ConnectionSemaphore,
}

impl ConnectOptions {
    /// Establish a TCP connection to the given URI and default port.
    async fn connect_tcp(
        &self,
        uri: &Uri,
        default_port: u16,
    ) -> Result<PermittedTcpStream, ErrorCode> {
        let mut socket_addrs = match self.override_connect_addr {
            Some(override_connect_addr) => vec![override_connect_addr],
            None => {
                let authority = uri.authority().ok_or(ErrorCode::HttpRequestUriInvalid)?;

                let host_and_port = if authority.port().is_some() {
                    authority.as_str().to_string()
                } else {
                    format!("{}:{}", authority.as_str(), default_port)
                };

                let socket_addrs = tokio::net::lookup_host(&host_and_port)
                    .await
                    .map_err(|err| {
                        tracing::debug!(?host_and_port, ?err, "Error resolving host");
                        dns_error("address not available".into(), 0)
                    })?
                    .collect::<Vec<_>>();
                tracing::debug!(?host_and_port, ?socket_addrs, "Resolved host");
                socket_addrs
            }
        };

        // Remove blocked IPs
        crate::remove_blocked_addrs(&self.blocked_networks, &mut socket_addrs)?;

        let connect = async {
            // If we're limiting concurrent outbound requests, acquire a permit
            let permit = self.semaphore.acquire().await;
            (TcpStream::connect(&*socket_addrs).await, permit)
        };

        // Make sure that the connect timeout applies to both acquiring the outbound request permit and establishing the TCP connection,
        // since acquiring the permit could potentially take a long time if there are many outbound requests happening.
        let (stream, permit) = timeout(self.connect_timeout, connect)
            .await
            .map_err(|_| ErrorCode::ConnectionTimeout)?;
        let permit = permit.map_err(|_| ErrorCode::ConnectionLimitReached)?;
        let stream = stream.map_err(|err| match err.kind() {
            std::io::ErrorKind::AddrNotAvailable => dns_error("address not available".into(), 0),
            _ => ErrorCode::ConnectionRefused,
        })?;
        Ok(PermittedTcpStream {
            inner: stream,
            _permit: permit,
        })
    }

    /// Establish a TLS connection to the given URI and default port.
    async fn connect_tls(
        &self,
        uri: &Uri,
        default_port: u16,
    ) -> Result<TlsStream<PermittedTcpStream>, ErrorCode> {
        let tcp_stream = self.connect_tcp(uri, default_port).await?;

        let mut tls_client_config = self.tls_client_config.as_deref().unwrap().clone();
        tls_client_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];

        let connector = tokio_rustls::TlsConnector::from(Arc::new(tls_client_config));
        let domain = rustls::pki_types::ServerName::try_from(uri.host().unwrap())
            .map_err(|e| {
                tracing::warn!("dns lookup error: {e:?}");
                dns_error("invalid dns name".into(), 0)
            })?
            .to_owned();
        connector.connect(domain, tcp_stream).await.map_err(|e| {
            tracing::warn!("tls protocol error: {e:?}");
            ErrorCode::TlsProtocolError
        })
    }
}

/// A connector the uses `ConnectOptions`
#[derive(Clone)]
struct HttpConnector;

impl HttpConnector {
    async fn connect(uri: Uri) -> Result<TokioIo<PermittedTcpStream>, ErrorCode> {
        let stream = CONNECT_OPTIONS.get().connect_tcp(&uri, 80).await?;
        Ok(TokioIo::new(stream))
    }
}

impl Service<Uri> for HttpConnector {
    type Response = TokioIo<PermittedTcpStream>;
    type Error = ErrorCode;
    type Future =
        Pin<Box<dyn Future<Output = Result<TokioIo<PermittedTcpStream>, ErrorCode>> + Send>>;

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

    fn call(&mut self, uri: Uri) -> Self::Future {
        Box::pin(async move { Self::connect(uri).await })
    }
}

/// A connector that establishes TLS connections using `rustls` and `ConnectOptions`.
#[derive(Clone)]
struct HttpsConnector;

impl HttpsConnector {
    async fn connect(uri: Uri) -> Result<TokioIo<RustlsStream>, ErrorCode> {
        let stream = CONNECT_OPTIONS.get().connect_tls(&uri, 443).await?;
        Ok(TokioIo::new(RustlsStream(stream)))
    }
}

impl Service<Uri> for HttpsConnector {
    type Response = TokioIo<RustlsStream>;
    type Error = ErrorCode;
    type Future = Pin<Box<dyn Future<Output = Result<TokioIo<RustlsStream>, ErrorCode>> + Send>>;

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

    fn call(&mut self, uri: Uri) -> Self::Future {
        Box::pin(async move { Self::connect(uri).await })
    }
}

struct RustlsStream(TlsStream<PermittedTcpStream>);

impl Connection for RustlsStream {
    fn connected(&self) -> Connected {
        if self.0.get_ref().1.alpn_protocol() == Some(b"h2") {
            self.0.get_ref().0.connected().negotiated_h2()
        } else {
            self.0.get_ref().0.connected()
        }
    }
}

impl AsyncRead for RustlsStream {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<Result<(), std::io::Error>> {
        Pin::new(&mut self.get_mut().0).poll_read(cx, buf)
    }
}

impl AsyncWrite for RustlsStream {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, std::io::Error>> {
        Pin::new(&mut self.get_mut().0).poll_write(cx, buf)
    }

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

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

    fn poll_write_vectored(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        bufs: &[IoSlice<'_>],
    ) -> Poll<Result<usize, std::io::Error>> {
        Pin::new(&mut self.get_mut().0).poll_write_vectored(cx, bufs)
    }

    fn is_write_vectored(&self) -> bool {
        self.0.is_write_vectored()
    }
}

/// A TCP stream that holds a permit indicating that it is allowed to exist.
struct PermittedTcpStream {
    /// The wrapped TCP stream.
    inner: TcpStream,
    /// A permit indicating that this stream is allowed to exist.
    ///
    /// When this stream is dropped, the permit is also dropped, allowing another
    /// connection to be established.
    _permit: ConnectionPermit,
}

impl Connection for PermittedTcpStream {
    fn connected(&self) -> Connected {
        self.inner.connected()
    }
}

impl AsyncRead for PermittedTcpStream {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.get_mut().inner).poll_read(cx, buf)
    }
}

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

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

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

/// Translate a [`hyper::Error`] to a wasi-http `ErrorCode` in the context of a request.
fn hyper_request_error(err: hyper::Error) -> ErrorCode {
    // If there's a source, we might be able to extract a wasi-http error from it.
    if let Some(cause) = err.source()
        && let Some(err) = cause.downcast_ref::<ErrorCode>()
    {
        return err.clone();
    }

    tracing::warn!("hyper request error: {err:?}");

    ErrorCode::HttpProtocolError
}

/// Translate a [`hyper_util::client::legacy::Error`] to a wasi-http `ErrorCode` in the context of a request.
fn hyper_legacy_request_error(err: hyper_util::client::legacy::Error) -> ErrorCode {
    // If there's a source, we might be able to extract a wasi-http error from it.
    if let Some(cause) = err.source()
        && let Some(err) = cause.downcast_ref::<ErrorCode>()
    {
        return err.clone();
    }

    tracing::warn!("hyper request error: {err:?}");

    ErrorCode::HttpProtocolError
}

fn dns_error(rcode: String, info_code: u16) -> ErrorCode {
    ErrorCode::DnsError(
        wasmtime_wasi_http::p2::bindings::http::types::DnsErrorPayload {
            rcode: Some(rcode),
            info_code: Some(info_code),
        },
    )
}

// TODO: Remove this (and uses of it) once
// https://github.com/spinframework/spin/issues/3274 has been addressed.
pub fn p2_to_p3_error_code(code: p2_types::ErrorCode) -> p3_types::ErrorCode {
    match code {
        p2_types::ErrorCode::DnsTimeout => p3_types::ErrorCode::DnsTimeout,
        p2_types::ErrorCode::DnsError(payload) => {
            p3_types::ErrorCode::DnsError(p3_types::DnsErrorPayload {
                rcode: payload.rcode,
                info_code: payload.info_code,
            })
        }
        p2_types::ErrorCode::DestinationNotFound => p3_types::ErrorCode::DestinationNotFound,
        p2_types::ErrorCode::DestinationUnavailable => p3_types::ErrorCode::DestinationUnavailable,
        p2_types::ErrorCode::DestinationIpProhibited => {
            p3_types::ErrorCode::DestinationIpProhibited
        }
        p2_types::ErrorCode::DestinationIpUnroutable => {
            p3_types::ErrorCode::DestinationIpUnroutable
        }
        p2_types::ErrorCode::ConnectionRefused => p3_types::ErrorCode::ConnectionRefused,
        p2_types::ErrorCode::ConnectionTerminated => p3_types::ErrorCode::ConnectionTerminated,
        p2_types::ErrorCode::ConnectionTimeout => p3_types::ErrorCode::ConnectionTimeout,
        p2_types::ErrorCode::ConnectionReadTimeout => p3_types::ErrorCode::ConnectionReadTimeout,
        p2_types::ErrorCode::ConnectionWriteTimeout => p3_types::ErrorCode::ConnectionWriteTimeout,
        p2_types::ErrorCode::ConnectionLimitReached => p3_types::ErrorCode::ConnectionLimitReached,
        p2_types::ErrorCode::TlsProtocolError => p3_types::ErrorCode::TlsProtocolError,
        p2_types::ErrorCode::TlsCertificateError => p3_types::ErrorCode::TlsCertificateError,
        p2_types::ErrorCode::TlsAlertReceived(payload) => {
            p3_types::ErrorCode::TlsAlertReceived(p3_types::TlsAlertReceivedPayload {
                alert_id: payload.alert_id,
                alert_message: payload.alert_message,
            })
        }
        p2_types::ErrorCode::HttpRequestDenied => p3_types::ErrorCode::HttpRequestDenied,
        p2_types::ErrorCode::HttpRequestLengthRequired => {
            p3_types::ErrorCode::HttpRequestLengthRequired
        }
        p2_types::ErrorCode::HttpRequestBodySize(payload) => {
            p3_types::ErrorCode::HttpRequestBodySize(payload)
        }
        p2_types::ErrorCode::HttpRequestMethodInvalid => {
            p3_types::ErrorCode::HttpRequestMethodInvalid
        }
        p2_types::ErrorCode::HttpRequestUriInvalid => p3_types::ErrorCode::HttpRequestUriInvalid,
        p2_types::ErrorCode::HttpRequestUriTooLong => p3_types::ErrorCode::HttpRequestUriTooLong,
        p2_types::ErrorCode::HttpRequestHeaderSectionSize(payload) => {
            p3_types::ErrorCode::HttpRequestHeaderSectionSize(payload)
        }
        p2_types::ErrorCode::HttpRequestHeaderSize(payload) => {
            p3_types::ErrorCode::HttpRequestHeaderSize(payload.map(|payload| {
                p3_types::FieldSizePayload {
                    field_name: payload.field_name,
                    field_size: payload.field_size,
                }
            }))
        }
        p2_types::ErrorCode::HttpRequestTrailerSectionSize(payload) => {
            p3_types::ErrorCode::HttpRequestTrailerSectionSize(payload)
        }
        p2_types::ErrorCode::HttpRequestTrailerSize(payload) => {
            p3_types::ErrorCode::HttpRequestTrailerSize(p3_types::FieldSizePayload {
                field_name: payload.field_name,
                field_size: payload.field_size,
            })
        }
        p2_types::ErrorCode::HttpResponseIncomplete => p3_types::ErrorCode::HttpResponseIncomplete,
        p2_types::ErrorCode::HttpResponseHeaderSectionSize(payload) => {
            p3_types::ErrorCode::HttpResponseHeaderSectionSize(payload)
        }
        p2_types::ErrorCode::HttpResponseHeaderSize(payload) => {
            p3_types::ErrorCode::HttpResponseHeaderSize(p3_types::FieldSizePayload {
                field_name: payload.field_name,
                field_size: payload.field_size,
            })
        }
        p2_types::ErrorCode::HttpResponseBodySize(payload) => {
            p3_types::ErrorCode::HttpResponseBodySize(payload)
        }
        p2_types::ErrorCode::HttpResponseTrailerSectionSize(payload) => {
            p3_types::ErrorCode::HttpResponseTrailerSectionSize(payload)
        }
        p2_types::ErrorCode::HttpResponseTrailerSize(payload) => {
            p3_types::ErrorCode::HttpResponseTrailerSize(p3_types::FieldSizePayload {
                field_name: payload.field_name,
                field_size: payload.field_size,
            })
        }
        p2_types::ErrorCode::HttpResponseTransferCoding(payload) => {
            p3_types::ErrorCode::HttpResponseTransferCoding(payload)
        }
        p2_types::ErrorCode::HttpResponseContentCoding(payload) => {
            p3_types::ErrorCode::HttpResponseContentCoding(payload)
        }
        p2_types::ErrorCode::HttpResponseTimeout => p3_types::ErrorCode::HttpResponseTimeout,
        p2_types::ErrorCode::HttpUpgradeFailed => p3_types::ErrorCode::HttpUpgradeFailed,
        p2_types::ErrorCode::HttpProtocolError => p3_types::ErrorCode::HttpProtocolError,
        p2_types::ErrorCode::LoopDetected => p3_types::ErrorCode::LoopDetected,
        p2_types::ErrorCode::ConfigurationError => p3_types::ErrorCode::ConfigurationError,
        p2_types::ErrorCode::InternalError(payload) => p3_types::ErrorCode::InternalError(payload),
    }
}

// TODO: Remove this (and uses of it) once
// https://github.com/spinframework/spin/issues/3274 has been addressed.
pub fn p3_to_p2_error_code(code: p3_types::ErrorCode) -> p2_types::ErrorCode {
    match code {
        p3_types::ErrorCode::DnsTimeout => p2_types::ErrorCode::DnsTimeout,
        p3_types::ErrorCode::DnsError(payload) => {
            p2_types::ErrorCode::DnsError(p2_types::DnsErrorPayload {
                rcode: payload.rcode,
                info_code: payload.info_code,
            })
        }
        p3_types::ErrorCode::DestinationNotFound => p2_types::ErrorCode::DestinationNotFound,
        p3_types::ErrorCode::DestinationUnavailable => p2_types::ErrorCode::DestinationUnavailable,
        p3_types::ErrorCode::DestinationIpProhibited => {
            p2_types::ErrorCode::DestinationIpProhibited
        }
        p3_types::ErrorCode::DestinationIpUnroutable => {
            p2_types::ErrorCode::DestinationIpUnroutable
        }
        p3_types::ErrorCode::ConnectionRefused => p2_types::ErrorCode::ConnectionRefused,
        p3_types::ErrorCode::ConnectionTerminated => p2_types::ErrorCode::ConnectionTerminated,
        p3_types::ErrorCode::ConnectionTimeout => p2_types::ErrorCode::ConnectionTimeout,
        p3_types::ErrorCode::ConnectionReadTimeout => p2_types::ErrorCode::ConnectionReadTimeout,
        p3_types::ErrorCode::ConnectionWriteTimeout => p2_types::ErrorCode::ConnectionWriteTimeout,
        p3_types::ErrorCode::ConnectionLimitReached => p2_types::ErrorCode::ConnectionLimitReached,
        p3_types::ErrorCode::TlsProtocolError => p2_types::ErrorCode::TlsProtocolError,
        p3_types::ErrorCode::TlsCertificateError => p2_types::ErrorCode::TlsCertificateError,
        p3_types::ErrorCode::TlsAlertReceived(payload) => {
            p2_types::ErrorCode::TlsAlertReceived(p2_types::TlsAlertReceivedPayload {
                alert_id: payload.alert_id,
                alert_message: payload.alert_message,
            })
        }
        p3_types::ErrorCode::HttpRequestDenied => p2_types::ErrorCode::HttpRequestDenied,
        p3_types::ErrorCode::HttpRequestLengthRequired => {
            p2_types::ErrorCode::HttpRequestLengthRequired
        }
        p3_types::ErrorCode::HttpRequestBodySize(payload) => {
            p2_types::ErrorCode::HttpRequestBodySize(payload)
        }
        p3_types::ErrorCode::HttpRequestMethodInvalid => {
            p2_types::ErrorCode::HttpRequestMethodInvalid
        }
        p3_types::ErrorCode::HttpRequestUriInvalid => p2_types::ErrorCode::HttpRequestUriInvalid,
        p3_types::ErrorCode::HttpRequestUriTooLong => p2_types::ErrorCode::HttpRequestUriTooLong,
        p3_types::ErrorCode::HttpRequestHeaderSectionSize(payload) => {
            p2_types::ErrorCode::HttpRequestHeaderSectionSize(payload)
        }
        p3_types::ErrorCode::HttpRequestHeaderSize(payload) => {
            p2_types::ErrorCode::HttpRequestHeaderSize(payload.map(|payload| {
                p2_types::FieldSizePayload {
                    field_name: payload.field_name,
                    field_size: payload.field_size,
                }
            }))
        }
        p3_types::ErrorCode::HttpRequestTrailerSectionSize(payload) => {
            p2_types::ErrorCode::HttpRequestTrailerSectionSize(payload)
        }
        p3_types::ErrorCode::HttpRequestTrailerSize(payload) => {
            p2_types::ErrorCode::HttpRequestTrailerSize(p2_types::FieldSizePayload {
                field_name: payload.field_name,
                field_size: payload.field_size,
            })
        }
        p3_types::ErrorCode::HttpResponseIncomplete => p2_types::ErrorCode::HttpResponseIncomplete,
        p3_types::ErrorCode::HttpResponseHeaderSectionSize(payload) => {
            p2_types::ErrorCode::HttpResponseHeaderSectionSize(payload)
        }
        p3_types::ErrorCode::HttpResponseHeaderSize(payload) => {
            p2_types::ErrorCode::HttpResponseHeaderSize(p2_types::FieldSizePayload {
                field_name: payload.field_name,
                field_size: payload.field_size,
            })
        }
        p3_types::ErrorCode::HttpResponseBodySize(payload) => {
            p2_types::ErrorCode::HttpResponseBodySize(payload)
        }
        p3_types::ErrorCode::HttpResponseTrailerSectionSize(payload) => {
            p2_types::ErrorCode::HttpResponseTrailerSectionSize(payload)
        }
        p3_types::ErrorCode::HttpResponseTrailerSize(payload) => {
            p2_types::ErrorCode::HttpResponseTrailerSize(p2_types::FieldSizePayload {
                field_name: payload.field_name,
                field_size: payload.field_size,
            })
        }
        p3_types::ErrorCode::HttpResponseTransferCoding(payload) => {
            p2_types::ErrorCode::HttpResponseTransferCoding(payload)
        }
        p3_types::ErrorCode::HttpResponseContentCoding(payload) => {
            p2_types::ErrorCode::HttpResponseContentCoding(payload)
        }
        p3_types::ErrorCode::HttpResponseTimeout => p2_types::ErrorCode::HttpResponseTimeout,
        p3_types::ErrorCode::HttpUpgradeFailed => p2_types::ErrorCode::HttpUpgradeFailed,
        p3_types::ErrorCode::HttpProtocolError => p2_types::ErrorCode::HttpProtocolError,
        p3_types::ErrorCode::LoopDetected => p2_types::ErrorCode::LoopDetected,
        p3_types::ErrorCode::ConfigurationError => p2_types::ErrorCode::ConfigurationError,
        p3_types::ErrorCode::InternalError(payload) => p2_types::ErrorCode::InternalError(payload),
    }
}

fn record_content_length_header(span: &Span, headers: &HeaderMap, attr_name: &'static str) {
    if let Some(content_length) = headers.get(CONTENT_LENGTH)
        && let Ok(size_str) = content_length.to_str()
    {
        span.set_attribute(attr_name, size_str.to_string());
    }
}

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

    /// Regression test: the connect timeout must cover permit acquisition, not
    /// just the TCP handshake.  Before the fix, a fully-saturated semaphore
    /// caused `connect_tcp` to hang indefinitely; after the fix it returns
    /// `ConnectionTimeout` within the configured deadline.
    #[tokio::test]
    async fn connect_timeout_applies_to_permit_acquisition() {
        // Create a semaphore with exactly 1 permit and immediately exhaust it, leaving
        // 0 permits available.  This simulates all outbound-connection slots being occupied.
        let conn_semaphore = ConnectionSemaphore::new(None, Some(1), "test", None);
        let _held = conn_semaphore
            .try_acquire()
            .expect("exhausting the single permit");

        let options = ConnectOptions {
            // No blocked networks; we want the address to pass the filter.
            blocked_networks: BlockedNetworks::default(),
            // A very short deadline so the test runs quickly.
            connect_timeout: Duration::from_millis(50),
            tls_client_config: None,
            // Skip DNS by supplying the address directly.
            override_connect_addr: Some("127.0.0.1:1".parse().unwrap()),
            semaphore: conn_semaphore,
        };

        // `connect_tcp` must time out while waiting for a permit rather than
        // blocking forever.
        let result = options
            .connect_tcp(&Uri::from_static("http://test.example"), 80)
            .await;

        assert!(
            matches!(result, Err(ErrorCode::ConnectionTimeout)),
            "expected ConnectionTimeout"
        );
    }
}