factor-outbound-http: Update override_connect_host

- Rename to override_connect_endpoint
- Support optionally overriding port by making the value an 'Authority'
- Fix bug in host resolution for ipv6 addresses

Signed-off-by: Lann Martin <[email protected]>

Author: lann

crates/factor-outbound-http/src/intercept.rs

@@ -3,6 +3,8 @@ use http_body_util::{BodyExt, Full};
 use spin_world::async_trait;
 use wasmtime_wasi_http::{body::HyperOutgoingBody, HttpResult};
 
+pub use http::uri::Authority;
+
 pub type HyperBody = HyperOutgoingBody;
 
 /// An outbound HTTP request interceptor to be used with
@@ -39,7 +41,7 @@ pub enum InterceptOutcome {
 pub struct InterceptRequest {
     inner: Request<()>,
     body: InterceptBody,
-    pub(crate) override_connect_host: Option<String>,
+    pub(crate) override_connect_endpoint: Option<Authority>,
 }
 
 enum InterceptBody {
@@ -48,16 +50,18 @@ enum InterceptBody {
 }
 
 impl InterceptRequest {
-    /// Overrides the host that will be connected to for this outbound request.
+    /// Overrides the endpoint (host and optional port) that will be connected
+    /// to for this outbound request.
     ///
     /// This override does not have any effect on TLS server name checking or
     /// HTTP authority / host headers.
     ///
-    /// This host will not be checked against `allowed_outbound_hosts`; if that
-    /// check should occur it must be performed by the interceptor. The resolved
-    /// IP addresses from this host will be checked against blocked IP networks.
-    pub fn override_connect_host(&mut self, host: impl Into<String>) {
-        self.override_connect_host = Some(host.into())
+    /// This endpoint will not be checked against `allowed_outbound_hosts`; if
+    /// that check should occur it must be performed by the interceptor. The
+    /// resolved IP addresses from this host will be checked against blocked IP
+    /// networks.
+    pub fn override_connect_endpoint(&mut self, endpoint: Authority) {
+        self.override_connect_endpoint = Some(endpoint);
     }
 
     pub fn into_hyper_request(self) -> Request<HyperBody> {
@@ -94,7 +98,7 @@ impl From<Request<HyperBody>> for InterceptRequest {
         Self {
             inner: Request::from_parts(parts, ()),
             body: InterceptBody::Hyper(body),
-            override_connect_host: None,
+            override_connect_endpoint: None,
         }
     }
 }
@@ -105,7 +109,7 @@ impl From<Request<Vec<u8>>> for InterceptRequest {
         Self {
             inner: Request::from_parts(parts, ()),
             body: InterceptBody::Vec(body),
-            override_connect_host: None,
+            override_connect_endpoint: None,
         }
     }
 }

crates/factor-outbound-http/src/wasi.rs

@@ -8,7 +8,7 @@ use std::{
     time::Duration,
 };
 
-use http::{header::HOST, Uri};
+use http::{header::HOST, uri::Authority, Uri};
 use http_body_util::BodyExt;
 use hyper_util::{
     client::legacy::{
@@ -113,7 +113,7 @@ impl WasiHttpView for WasiHttpImplInner<'_> {
             http.response.status_code = Empty,
             server.address = Empty,
             server.port = Empty,
-        ),
+        )
     )]
     fn send_request(
         &mut self,
@@ -166,12 +166,12 @@ impl RequestSender {
         spin_telemetry::inject_trace_context(&mut request);
 
         // Run any configured request interceptor
-        let mut override_connect_host = None;
+        let mut override_connect_endpoint = 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_host = req.override_connect_host.take();
+                    override_connect_endpoint = req.override_connect_endpoint.take();
                     request = req.into_hyper_request();
                 }
                 InterceptOutcome::Complete(resp) => {
@@ -188,15 +188,15 @@ impl RequestSender {
         // Backfill span fields after potentially updating the URL in the interceptor
         if let Some(authority) = request.uri().authority() {
             let span = tracing::Span::current();
-            let host = override_connect_host.as_deref().unwrap_or(authority.host());
-            span.record("server.address", host);
-            if let Some(port) = authority.port() {
-                span.record("server.port", port.as_u16());
+            let authority = override_connect_endpoint.as_ref().unwrap_or(authority);
+            span.record("server.address", authority.host());
+            if let Some(port) = authority.port_u16() {
+                span.record("server.port", port);
             }
         }
 
         Ok(self
-            .send_request(request, config, override_connect_host)
+            .send_request(request, config, override_connect_endpoint)
             .await?)
     }
 
@@ -275,7 +275,7 @@ impl RequestSender {
         self,
         request: OutgoingRequest,
         config: OutgoingRequestConfig,
-        override_connect_host: Option<String>,
+        override_connect_endpoint: Option<Authority>,
     ) -> Result<IncomingResponse, ErrorCode> {
         let OutgoingRequestConfig {
             use_tls,
@@ -296,7 +296,7 @@ impl RequestSender {
                 blocked_networks: self.blocked_networks,
                 connect_timeout,
                 tls_client_config,
-                override_connect_host,
+                override_connect_endpoint,
             },
             async move {
                 if use_tls {
@@ -376,19 +376,24 @@ struct ConnectOptions {
     blocked_networks: BlockedNetworks,
     connect_timeout: Duration,
     tls_client_config: Option<TlsClientConfig>,
-    override_connect_host: Option<String>,
+    override_connect_endpoint: Option<Authority>,
 }
 
 impl ConnectOptions {
     async fn connect_tcp(&self, uri: &Uri, default_port: u16) -> Result<TcpStream, ErrorCode> {
-        let host = self
-            .override_connect_host
-            .as_deref()
-            .or(uri.host())
+        let authority = self
+            .override_connect_endpoint
+            .as_ref()
+            .or(uri.authority())
             .ok_or(ErrorCode::HttpRequestUriInvalid)?;
-        let host_and_port = (host, uri.port_u16().unwrap_or(default_port));
 
-        let mut socket_addrs = tokio::net::lookup_host(host_and_port)
+        let host_and_port = if authority.port().is_some() {
+            authority.as_str().to_string()
+        } else {
+            format!("{}:{}", authority.as_str(), default_port)
+        };
+
+        let mut socket_addrs = tokio::net::lookup_host(&host_and_port)
             .await
             .map_err(|err| {
                 tracing::debug!(?host_and_port, ?err, "Error resolving host");

crates/factor-outbound-http/tests/factor_test.rs

@@ -22,7 +22,7 @@ struct TestFactors {
     http: OutboundHttpFactor,
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn allowed_host_is_allowed() -> anyhow::Result<()> {
     let mut state = test_instance_state("https://*", true).await?;
     let mut wasi_http = OutboundHttpFactor::get_wasi_http_impl(&mut state).unwrap();
@@ -36,7 +36,7 @@ async fn allowed_host_is_allowed() -> anyhow::Result<()> {
     Ok(())
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn self_request_smoke_test() -> anyhow::Result<()> {
     let mut state = test_instance_state("http://self", true).await?;
     // [100::] is the IPv6 "Discard Prefix", which should always fail
@@ -52,7 +52,7 @@ async fn self_request_smoke_test() -> anyhow::Result<()> {
     Ok(())
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn disallowed_host_fails() -> anyhow::Result<()> {
     let mut state = test_instance_state("https://allowed.test", true).await?;
     let mut wasi_http = OutboundHttpFactor::get_wasi_http_impl(&mut state).unwrap();
@@ -67,7 +67,7 @@ async fn disallowed_host_fails() -> anyhow::Result<()> {
     Ok(())
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn disallowed_private_ips_fails() -> anyhow::Result<()> {
     async fn run_test(allow_private_ips: bool) -> anyhow::Result<()> {
         let mut state = test_instance_state("http://*", allow_private_ips).await?;
@@ -100,8 +100,8 @@ async fn disallowed_private_ips_fails() -> anyhow::Result<()> {
     Ok(())
 }
 
-#[tokio::test]
-async fn override_connect_host_disallowed_private_ip_fails() -> anyhow::Result<()> {
+#[tokio::test(flavor = "multi_thread")]
+async fn override_connect_endpoint_disallowed_private_ip_fails() -> anyhow::Result<()> {
     let mut state = test_instance_state("http://*", false).await?;
     state.http.set_request_interceptor({
         struct Interceptor;
@@ -111,7 +111,7 @@ async fn override_connect_host_disallowed_private_ip_fails() -> anyhow::Result<(
                 &self,
                 mut request: InterceptRequest,
             ) -> wasmtime_wasi_http::HttpResult<InterceptOutcome> {
-                request.override_connect_host("localhost");
+                request.override_connect_endpoint("[::1]".try_into().unwrap());
                 Ok(InterceptOutcome::Continue(request))
             }
         }
@@ -159,8 +159,8 @@ fn test_request_config() -> OutgoingRequestConfig {
     OutgoingRequestConfig {
         use_tls: false,
         connect_timeout: Duration::from_millis(10),
-        first_byte_timeout: Duration::from_millis(10),
-        between_bytes_timeout: Duration::from_millis(10),
+        first_byte_timeout: Duration::from_millis(0),
+        between_bytes_timeout: Duration::from_millis(0),
     }
 }