Share buffered AWC transport for OIDC and telemetry

src/telemetry.rs

@@ -9,6 +9,7 @@
 use std::env;
 use std::sync::{Once, OnceLock};
 
+use crate::webserver::http_client::send_http_request;
 use anyhow::Context as _;
 use opentelemetry_sdk::metrics::SdkMeterProvider;
 use opentelemetry_sdk::trace::SdkTracerProvider;
@@ -90,9 +91,9 @@ fn init_otlp_http_worker_sender() -> anyhow::Result<tokio::sync::mpsc::Unbounded
                     };
 
                 while let Some(job) = receiver.recv().await {
-                    let response = execute_otlp_http_request_with_awc(&awc_client, job.request)
+                    let response = send_http_request(&awc_client, job.request, None)
                         .await
-                        .map_err(|error| error.to_string());
+                        .map_err(|error| format!("{error:#}"));
                     let _ = job.response_sender.send(response);
                 }
             });
@@ -108,55 +109,6 @@ fn init_otlp_http_worker_sender() -> anyhow::Result<tokio::sync::mpsc::Unbounded
     Ok(sender)
 }
 
-async fn execute_otlp_http_request_with_awc(
-    awc_client: &awc::Client,
-    request: opentelemetry_http::Request<opentelemetry_http::Bytes>,
-) -> anyhow::Result<opentelemetry_http::Response<opentelemetry_http::Bytes>> {
-    let (request_parts, request_body) = request.into_parts();
-
-    let awc_method = awc::http::Method::from_bytes(request_parts.method.as_str().as_bytes())
-        .with_context(|| format!("Invalid OTLP HTTP method: {}", request_parts.method))?;
-    let awc_uri: awc::http::Uri = request_parts
-        .uri
-        .to_string()
-        .parse()
-        .with_context(|| format!("Invalid OTLP collector URI: {}", request_parts.uri))?;
-
-    let mut awc_request = awc_client.request(awc_method, awc_uri.clone());
-    for (header_name, header_value) in &request_parts.headers {
-        let header_name_str = header_name.as_str();
-        let awc_header_name = awc::http::header::HeaderName::from_bytes(header_name_str.as_bytes())
-            .with_context(|| format!("Invalid OTLP header name: {header_name_str}"))?;
-        let awc_header_value = awc::http::header::HeaderValue::from_bytes(header_value.as_bytes())
-            .with_context(|| format!("Invalid OTLP header value for {header_name_str}"))?;
-        awc_request = awc_request.insert_header((awc_header_name, awc_header_value));
-    }
-
-    let mut awc_response = awc_request.send_body(request_body).await.map_err(|error| {
-        anyhow::anyhow!("Failed to send OTLP HTTP request to {awc_uri}: {error}")
-    })?;
-
-    let mut response_builder =
-        opentelemetry_http::Response::builder().status(awc_response.status().as_u16());
-    for (header_name, header_value) in awc_response.headers() {
-        let header_value = header_value.to_str().map_err(|error| {
-            anyhow::anyhow!(
-                "Invalid OTLP response header value for {}: {error}",
-                header_name.as_str()
-            )
-        })?;
-        response_builder = response_builder.header(header_name.as_str(), header_value);
-    }
-
-    let response_body = awc_response.body().await.map_err(|error| {
-        anyhow::anyhow!("Failed to read OTLP HTTP response body from {awc_uri}: {error}")
-    })?;
-
-    response_builder
-        .body(response_body)
-        .context("Failed to build OTLP HTTP response")
-}
-
 #[async_trait::async_trait]
 impl opentelemetry_http::HttpClient for AwcOtlpHttpClient {
     async fn send_bytes(

src/webserver/http_client.rs

@@ -1,7 +1,54 @@
-use actix_web::dev::ServiceRequest;
+use actix_web::{dev::ServiceRequest, web::Bytes};
 use anyhow::{Context, anyhow};
+use opentelemetry_http::{Request, Response};
 use rustls_native_certs::CertificateResult;
 use std::sync::OnceLock;
+use std::time::Duration;
+
+/// Bridges the HTTP 1.x types used by OIDC and OpenTelemetry to AWC's HTTP 0.2 types.
+/// Validates response headers before buffering the body.
+pub(crate) async fn send_http_request<B>(
+    client: &awc::Client,
+    request: Request<B>,
+    body_timeout: Option<Duration>,
+) -> anyhow::Result<Response<Bytes>>
+where
+    B: actix_web::body::MessageBody + 'static,
+{
+    let (head, body) = request.into_parts();
+    let method = awc::http::Method::from_bytes(head.method.as_str().as_bytes())?;
+    let mut request = client.request(method, head.uri.to_string());
+    for (name, value) in &head.headers {
+        request = request.insert_header((name.as_str(), value.as_bytes()));
+    }
+    log::debug!("Executing HTTP request: {} {}", head.method, head.uri);
+    let mut response = request.send_body(body).await.map_err(|error| {
+        anyhow!(
+            "Failed to send HTTP request: {} {}: {error}",
+            head.method,
+            head.uri
+        )
+    })?;
+    log::debug!("Received HTTP response: {}", response.status());
+    let mut builder = Response::builder().status(response.status().as_u16());
+    for (name, value) in response.headers() {
+        builder = builder.header(
+            name.as_str(),
+            value
+                .to_str()
+                .with_context(|| format!("Invalid HTTP response header: {name}"))?,
+        );
+    }
+    if let Some(timeout) = body_timeout {
+        response = response.timeout(timeout);
+    }
+    let body = response
+        .body()
+        .await
+        .with_context(|| format!("Failed to read HTTP response body from {}", head.uri))?;
+    log::debug!("Received HTTP response body: {} bytes", body.len());
+    Ok(builder.body(body)?)
+}
 
 struct NativeCertificates {
     certificates: Vec<rustls::pki_types::CertificateDer<'static>>,
@@ -101,3 +148,183 @@ pub(crate) fn get_http_client_from_appdata(
         Err(anyhow!("HTTP client not found in app data"))
     }
 }
+
+#[cfg(test)]
+mod transport_tests {
+    use super::*;
+    use openidconnect::{AsyncHttpClient, http::HeaderValue};
+    use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
+    use tokio::net::TcpListener;
+    use tokio::task::JoinHandle;
+
+    #[derive(Clone, Copy, Debug)]
+    enum Adapter {
+        Oidc,
+        Otlp,
+    }
+    use Adapter::{Oidc, Otlp};
+
+    impl Adapter {
+        async fn request(
+            self,
+            url: &str,
+            method: &str,
+            header: &[u8],
+        ) -> anyhow::Result<Response<Vec<u8>>> {
+            let request = Request::builder()
+                .uri(url)
+                .method(method)
+                .header("x-request", header)
+                .body(b"request".to_vec())?;
+            let client = awc::Client::default();
+            match self {
+                Oidc => super::super::oidc::AwcHttpClient::from_client(&client)
+                    .call(request)
+                    .await
+                    .map_err(anyhow::Error::msg),
+                Otlp => send_http_request(&client, request.map(Bytes::from), None)
+                    .await
+                    .map(|response| response.map(|body| body.to_vec())),
+            }
+        }
+    }
+
+    // Raw responses are needed for invalid header bytes, invalid chunks, and stalled bodies.
+    async fn response_server(
+        response: &'static [u8],
+        hold_open: bool,
+    ) -> (String, JoinHandle<Vec<u8>>) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("http://{}/transport", listener.local_addr().unwrap());
+        let task = tokio::spawn(async move {
+            let (socket, _) = listener.accept().await.unwrap();
+            let mut socket = BufReader::new(socket);
+            let mut request = Vec::new();
+            while !request.ends_with(b"\r\n\r\n") {
+                assert!(socket.read_until(b'\n', &mut request).await.unwrap() > 0);
+            }
+            let length = String::from_utf8_lossy(&request)
+                .lines()
+                .filter_map(|line| line.split_once(':'))
+                .find(|(name, _)| name.eq_ignore_ascii_case("content-length"))
+                .map_or(0, |(_, value)| value.trim().parse::<usize>().unwrap());
+            let body_start = request.len();
+            request.resize(body_start + length, 0);
+            socket.read_exact(&mut request[body_start..]).await.unwrap();
+            socket.get_mut().write_all(response).await.unwrap();
+            if hold_open {
+                std::future::pending::<()>().await;
+            }
+            request
+        });
+        (url, task)
+    }
+
+    #[actix_web::test]
+    async fn preserves_methods_bodies_status_and_headers() -> anyhow::Result<()> {
+        for adapter in [Oidc, Otlp] {
+            for method in ["GET", "POST", "PATCH", "DELETE"] {
+                let (url, server) = response_server(
+                    b"HTTP/1.1 202 Accepted\r\nContent-Length: 8\r\nX-Result: first\r\nX-Result: second\r\n\r\nresponse", false,
+                ).await;
+                let header = match adapter {
+                    Oidc => b"value".as_slice(),
+                    Otlp => b"\xff".as_slice(),
+                };
+                let response = adapter.request(&url, method, header).await?;
+                assert_eq!(response.status().as_u16(), 202);
+                assert_eq!(response.body(), b"response");
+                assert_eq!(response.headers().get_all("x-result").iter().count(), 2);
+                let sent = server.await?;
+                assert!(sent.starts_with(format!("{method} /transport HTTP/1.1\r\n").as_bytes()));
+                let expected_header = [b"x-request: ".as_slice(), header, b"\r\n"].concat();
+                assert!(
+                    sent.windows(expected_header.len())
+                        .any(|bytes| bytes == expected_header)
+                );
+                assert!(sent.ends_with(b"\r\n\r\nrequest"));
+            }
+        }
+        Ok(())
+    }
+
+    #[actix_web::test]
+    async fn rejects_non_text_headers_before_reading_body() -> anyhow::Result<()> {
+        let error = Oidc
+            .request("http://127.0.0.1:1/transport", "GET", b"\xff")
+            .await
+            .unwrap_err();
+        assert_eq!(
+            error.to_string(),
+            HeaderValue::from_bytes(b"\xff")?
+                .to_str()
+                .unwrap_err()
+                .to_string()
+        );
+        for adapter in [Oidc, Otlp] {
+            let (url, server) = response_server(
+                b"HTTP/1.1 200 OK\r\nX-Invalid: \xff\r\nContent-Length: 100\r\n\r\n",
+                true,
+            )
+            .await;
+            let error = tokio::time::timeout(
+                Duration::from_secs(1),
+                adapter.request(&url, "GET", b"value"),
+            )
+            .await?
+            .unwrap_err();
+            assert!(
+                error
+                    .to_string()
+                    .starts_with("Invalid HTTP response header: x-invalid")
+            );
+            server.abort();
+        }
+        Ok(())
+    }
+
+    #[actix_web::test]
+    async fn retains_send_and_body_error_context() -> anyhow::Result<()> {
+        for adapter in [Oidc, Otlp] {
+            let listener = TcpListener::bind("127.0.0.1:0").await?;
+            let unavailable = format!("http://{}/transport", listener.local_addr()?);
+            drop(listener);
+            let error = adapter
+                .request(&unavailable, "GET", b"value")
+                .await
+                .unwrap_err();
+            assert!(
+                error
+                    .to_string()
+                    .starts_with("Failed to send HTTP request: GET ")
+            );
+            let (url, server) = response_server(
+                b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\ninvalid\r\n",
+                false,
+            )
+            .await;
+            let error = adapter.request(&url, "GET", b"value").await.unwrap_err();
+            assert_eq!(
+                error.to_string(),
+                format!("Failed to read HTTP response body from {url}")
+            );
+            server.await?;
+        }
+        Ok(())
+    }
+
+    #[actix_web::test]
+    async fn applies_body_timeout_after_validating_headers() {
+        let (url, server) =
+            response_server(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n", true).await;
+        let error = send_http_request(
+            &awc::Client::default(),
+            Request::builder().uri(&url).body(Vec::<u8>::new()).unwrap(),
+            Some(Duration::from_millis(20)),
+        )
+        .await
+        .unwrap_err();
+        assert!(error.downcast_ref::<awc::error::PayloadError>().is_some());
+        server.abort();
+    }
+}

src/webserver/oidc.rs

@@ -5,7 +5,7 @@ use std::time::Duration;
 use std::{future::Future, pin::Pin, str::FromStr, sync::Arc};
 use tokio::time::Instant;
 
-use crate::webserver::http_client::get_http_client_from_appdata;
+use crate::webserver::http_client::{get_http_client_from_appdata, send_http_request};
 use crate::webserver::routing::{
     AppFileStore, CanonicalRequestPath, RoutingAction, canonical_url_path, resolve_route,
 };
@@ -1022,55 +1022,21 @@ impl<'c> AsyncHttpClient<'c> for AwcHttpClient<'c> {
         Pin<Box<dyn Future<Output = Result<openidconnect::HttpResponse, Self::Error>> + 'c>>;
 
     fn call(&'c self, request: openidconnect::HttpRequest) -> Self::Future {
-        let client = self.client.clone();
         Box::pin(async move {
-            execute_oidc_request_with_awc(client, request)
+            // OIDC accepts only text request headers; OTLP also accepts opaque bytes.
+            for value in request.headers().values() {
+                value
+                    .to_str()
+                    .map_err(|error| AwcWrapperError(error.into()))?;
+            }
+            send_http_request(self.client, request, Some(OIDC_HTTP_BODY_TIMEOUT))
                 .await
+                .map(|response| response.map(|body| body.to_vec()))
                 .map_err(AwcWrapperError)
         })
     }
 }
 
-async fn execute_oidc_request_with_awc(
-    client: Client,
-    request: openidconnect::HttpRequest,
-) -> Result<openidconnect::http::Response<Vec<u8>>, anyhow::Error> {
-    let awc_method = awc::http::Method::from_bytes(request.method().as_str().as_bytes())?;
-    let awc_uri = awc::http::Uri::from_str(&request.uri().to_string())?;
-    log::debug!("Executing OIDC request: {awc_method} {awc_uri}");
-    let mut req = client.request(awc_method, awc_uri);
-    for (name, value) in request.headers() {
-        req = req.insert_header((name.as_str(), value.to_str()?));
-    }
-    let (req_head, body) = request.into_parts();
-    let response = req.send_body(body).await.map_err(|e| {
-        anyhow!(e.to_string()).context(format!(
-            "Failed to send request: {} {}",
-            req_head.method, req_head.uri
-        ))
-    })?;
-    let head = response.headers();
-    log::debug!(
-        "Received OIDC response headers: status={}, content_type={:?}",
-        response.status(),
-        head.get(header::CONTENT_TYPE)
-            .and_then(|value| value.to_str().ok())
-    );
-    let mut resp_builder =
-        openidconnect::http::Response::builder().status(response.status().as_u16());
-    for (name, value) in head {
-        resp_builder = resp_builder.header(name.as_str(), value.to_str()?);
-    }
-    let mut response = response.timeout(OIDC_HTTP_BODY_TIMEOUT);
-    let body = response
-        .body()
-        .await
-        .with_context(|| format!("Couldnt read from {}", req_head.uri))?;
-    log::debug!("Received OIDC response body_len={} bytes", body.len());
-    let resp = resp_builder.body(body.to_vec())?;
-    Ok(resp)
-}
-
 #[derive(Debug)]
 pub struct AwcWrapperError(anyhow::Error);