From e6f81e70ef9a10d00919b6cf8846b77b05f3b224 Mon Sep 17 00:00:00 2001 From: Ophir Lojkine Date: Wed, 7 Oct 2026 09:52:01 +0000 Subject: [PATCH 1/3] Share buffered AWC transport for OIDC and telemetry --- src/telemetry.rs | 134 +++++++++++++++++++++++++----- src/webserver/http_client.rs | 107 ++++++++++++++++++++++++ src/webserver/oidc.rs | 152 ++++++++++++++++++++++++++++++----- 3 files changed, 354 insertions(+), 39 deletions(-) diff --git a/src/telemetry.rs b/src/telemetry.rs index 41c669edb..471ee97d8 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -9,6 +9,7 @@ use std::env; use std::sync::{Once, OnceLock}; +use crate::webserver::http_client::{BufferedRequestError, send_buffered_request}; use anyhow::Context as _; use opentelemetry_sdk::metrics::SdkMeterProvider; use opentelemetry_sdk::trace::SdkTracerProvider; @@ -132,25 +133,30 @@ async fn execute_otlp_http_request_with_awc( 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() - ) + let (response_builder, response_body) = + send_buffered_request(awc_request, request_body, None, |status, headers| { + let mut builder = opentelemetry_http::Response::builder().status(status.as_u16()); + for (header_name, header_value) in headers { + let header_value = header_value.to_str().map_err(|error| { + anyhow::anyhow!( + "Invalid OTLP response header value for {}: {error}", + header_name.as_str() + ) + })?; + builder = builder.header(header_name.as_str(), header_value); + } + Ok(builder) + }) + .await + .map_err(|error| match error { + BufferedRequestError::Send(error) => { + anyhow::anyhow!("Failed to send OTLP HTTP request to {awc_uri}: {error}") + } + BufferedRequestError::ResponseHead(error) => error, + BufferedRequestError::Body(error) => { + anyhow::anyhow!("Failed to read OTLP HTTP response body from {awc_uri}: {error}") + } })?; - 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) @@ -181,6 +187,98 @@ impl opentelemetry_http::HttpClient for AwcOtlpHttpClient { } } +#[cfg(test)] +mod http_transport_tests { + use super::*; + use crate::webserver::http_client::transport_tests::{response_server, unavailable_url}; + use std::time::Duration; + + #[actix_web::test] + async fn preserves_methods_bodies_status_and_raw_request_headers() -> anyhow::Result<()> { + 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 request = opentelemetry_http::Request::builder() + .method(method) + .uri(url) + .header("x-request", b"\xff".as_slice()) + .body(opentelemetry_http::Bytes::from_static(b"request"))?; + let response = + execute_otlp_http_request_with_awc(&awc::Client::default(), request).await?; + assert_eq!(response.status().as_u16(), 202); + assert_eq!(response.body().as_ref(), 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())); + assert!( + sent.windows(14) + .any(|bytes| bytes == b"x-request: \xff\r\n") + ); + assert!(sent.ends_with(b"\r\n\r\nrequest")); + } + Ok(()) + } + + #[actix_web::test] + async fn rejects_non_text_response_headers_before_reading_body() -> anyhow::Result<()> { + 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 request = opentelemetry_http::Request::builder() + .uri(url) + .body(opentelemetry_http::Bytes::new())?; + let error = tokio::time::timeout( + Duration::from_secs(1), + execute_otlp_http_request_with_awc(&awc::Client::default(), request), + ) + .await? + .unwrap_err(); + assert!( + error + .to_string() + .starts_with("Invalid OTLP response header value for x-invalid:") + ); + server.abort(); + Ok(()) + } + + #[actix_web::test] + async fn retains_send_and_body_error_context() -> anyhow::Result<()> { + let request = opentelemetry_http::Request::builder() + .uri(unavailable_url().await) + .body(opentelemetry_http::Bytes::new())?; + let error = execute_otlp_http_request_with_awc(&awc::Client::default(), request) + .await + .unwrap_err(); + assert!( + error + .to_string() + .starts_with("Failed to send OTLP HTTP request to ") + ); + + let (url, server) = response_server( + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\ninvalid\r\n", + false, + ) + .await; + let request = opentelemetry_http::Request::builder() + .uri(&url) + .body(opentelemetry_http::Bytes::new())?; + let error = execute_otlp_http_request_with_awc(&awc::Client::default(), request) + .await + .unwrap_err(); + assert!(error.to_string().starts_with(&format!( + "Failed to read OTLP HTTP response body from {url}:" + ))); + server.await?; + Ok(()) + } +} + /// Initializes logging / tracing. Returns whether `OTel` was activated. pub fn init_telemetry() -> anyhow::Result { init_telemetry_with_log_layer(logfmt::LogfmtLayer::new()) diff --git a/src/webserver/http_client.rs b/src/webserver/http_client.rs index 1f934b705..a064527c1 100644 --- a/src/webserver/http_client.rs +++ b/src/webserver/http_client.rs @@ -2,6 +2,43 @@ use actix_web::dev::ServiceRequest; use anyhow::{Context, anyhow}; use rustls_native_certs::CertificateResult; use std::sync::OnceLock; +use std::time::Duration; + +/// The stage that failed while buffering an outbound HTTP response. +#[derive(Debug)] +pub(crate) enum BufferedRequestError { + Send(awc::error::SendRequestError), + ResponseHead(anyhow::Error), + Body(awc::error::PayloadError), +} + +/// Sends a prepared request and validates the response head before reading its body. +/// Callers retain their header policy, timeout selection, and error context. +pub(crate) async fn send_buffered_request( + request: awc::ClientRequest, + body: B, + body_timeout: Option, + prepare_response_head: impl FnOnce( + awc::http::StatusCode, + &awc::http::header::HeaderMap, + ) -> anyhow::Result, +) -> Result<(H, actix_web::web::Bytes), BufferedRequestError> +where + B: actix_web::body::MessageBody + 'static, +{ + let response = request + .send_body(body) + .await + .map_err(BufferedRequestError::Send)?; + let head = prepare_response_head(response.status(), response.headers()) + .map_err(BufferedRequestError::ResponseHead)?; + let mut response = match body_timeout { + Some(timeout) => response.timeout(timeout), + None => response, + }; + let body = response.body().await.map_err(BufferedRequestError::Body)?; + Ok((head, body)) +} struct NativeCertificates { certificates: Vec>, @@ -101,3 +138,73 @@ pub(crate) fn get_http_client_from_appdata( Err(anyhow!("HTTP client not found in app data")) } } + +#[cfg(test)] +pub(crate) mod transport_tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + use tokio::task::JoinHandle; + + pub(crate) async fn response_server( + response: &'static [u8], + hold_open: bool, + ) -> (String, JoinHandle>) { + 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 (mut socket, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + let (header_end, content_length) = loop { + let mut buffer = [0; 1024]; + let count = socket.read(&mut buffer).await.unwrap(); + assert!(count > 0, "request ended before its headers"); + request.extend_from_slice(&buffer[..count]); + if let Some(end) = request.windows(4).position(|bytes| bytes == b"\r\n\r\n") { + let length = String::from_utf8_lossy(&request[..end]) + .lines() + .filter_map(|line| line.split_once(':')) + .find(|(name, _)| name.eq_ignore_ascii_case("content-length")) + .map_or(0, |(_, value)| value.trim().parse::().unwrap()); + break (end + 4, length); + } + }; + while request.len() < header_end + content_length { + let mut buffer = [0; 1024]; + let count = socket.read(&mut buffer).await.unwrap(); + assert!(count > 0, "request ended before its body"); + request.extend_from_slice(&buffer[..count]); + } + socket.write_all(response).await.unwrap(); + if hold_open { + std::future::pending::<()>().await; + } + request + }); + (url, task) + } + + pub(crate) async fn unavailable_url() -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + format!("http://{}/transport", listener.local_addr().unwrap()) + } + + #[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_buffered_request( + awc::Client::default().get(url), + Vec::::new(), + Some(Duration::from_millis(20)), + |status, _| { + assert_eq!(status, awc::http::StatusCode::OK); + Ok(()) + }, + ) + .await + .unwrap_err(); + assert!(matches!(error, BufferedRequestError::Body(_))); + server.abort(); + } +} diff --git a/src/webserver/oidc.rs b/src/webserver/oidc.rs index 153284894..a25dfa7d4 100644 --- a/src/webserver/oidc.rs +++ b/src/webserver/oidc.rs @@ -5,7 +5,9 @@ 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::{ + BufferedRequestError, get_http_client_from_appdata, send_buffered_request, +}; use crate::webserver::routing::{ AppFileStore, CanonicalRequestPath, RoutingAction, canonical_url_path, resolve_route, }; @@ -1043,29 +1045,36 @@ async fn execute_oidc_request_with_awc( 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!( + let (resp_builder, body) = send_buffered_request( + req, + body, + Some(OIDC_HTTP_BODY_TIMEOUT), + |status, headers| { + log::debug!( + "Received OIDC response headers: status={}, content_type={:?}", + status, + headers + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + ); + let mut builder = openidconnect::http::Response::builder().status(status.as_u16()); + for (name, value) in headers { + builder = builder.header(name.as_str(), value.to_str()?); + } + Ok(builder) + }, + ) + .await + .map_err(|error| match error { + BufferedRequestError::Send(error) => anyhow!(error.to_string()).context(format!( "Failed to send request: {} {}", req_head.method, req_head.uri - )) + )), + BufferedRequestError::ResponseHead(error) => error, + BufferedRequestError::Body(error) => { + anyhow::Error::new(error).context(format!("Couldnt read from {}", 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) @@ -1390,6 +1399,107 @@ fn validate_redirect_url(url: String, redirect_uri: &str) -> String { '/'.to_string() } +#[cfg(test)] +mod http_transport_tests { + use super::*; + use crate::webserver::http_client::transport_tests::{response_server, unavailable_url}; + + #[actix_web::test] + async fn preserves_methods_bodies_status_and_headers() -> anyhow::Result<()> { + 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 request = openidconnect::http::Request::builder() + .method(method) + .uri(url) + .header("x-request", "value") + .body(b"request".to_vec())?; + let response = execute_oidc_request_with_awc(Client::default(), request).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 = String::from_utf8(server.await?)?; + assert!(sent.starts_with(&format!("{method} /transport HTTP/1.1\r\n"))); + assert!(sent.contains("x-request: value\r\n")); + assert!(sent.ends_with("\r\n\r\nrequest")); + } + Ok(()) + } + + #[actix_web::test] + async fn rejects_non_text_request_and_response_headers() -> anyhow::Result<()> { + let request = openidconnect::http::Request::builder() + .uri("http://127.0.0.1:1/transport") + .header( + "x-invalid", + openidconnect::http::HeaderValue::from_bytes(b"\xff")?, + ) + .body(Vec::new())?; + let error = execute_oidc_request_with_awc(Client::default(), request) + .await + .unwrap_err(); + assert!( + error + .downcast_ref::() + .is_some() + ); + + 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 request = openidconnect::http::Request::builder() + .uri(url) + .body(Vec::new())?; + let error = tokio::time::timeout( + Duration::from_secs(1), + execute_oidc_request_with_awc(Client::default(), request), + ) + .await? + .unwrap_err(); + assert!( + error + .downcast_ref::() + .is_some() + ); + server.abort(); + Ok(()) + } + + #[actix_web::test] + async fn retains_send_and_body_error_context() -> anyhow::Result<()> { + let request = openidconnect::http::Request::builder() + .uri(unavailable_url().await) + .body(Vec::new())?; + let error = execute_oidc_request_with_awc(Client::default(), request) + .await + .unwrap_err(); + assert!( + error + .to_string() + .starts_with("Failed to send 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 request = openidconnect::http::Request::builder() + .uri(&url) + .body(Vec::new())?; + let error = execute_oidc_request_with_awc(Client::default(), request) + .await + .unwrap_err(); + assert_eq!(error.to_string(), format!("Couldnt read from {url}")); + server.await?; + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; From 8dd9376a2db97e92a3f4e22f172e509e18f5eca0 Mon Sep 17 00:00:00 2001 From: Ophir Lojkine Date: Wed, 7 Oct 2026 10:06:49 +0000 Subject: [PATCH 2/3] Consolidate HTTP adapter regression coverage --- src/telemetry.rs | 94 +------------------ src/webserver/http_client.rs | 170 +++++++++++++++++++++++++++++------ src/webserver/oidc.rs | 103 +-------------------- 3 files changed, 145 insertions(+), 222 deletions(-) diff --git a/src/telemetry.rs b/src/telemetry.rs index 471ee97d8..b8a337198 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -109,7 +109,7 @@ fn init_otlp_http_worker_sender() -> anyhow::Result, ) -> anyhow::Result> { @@ -187,98 +187,6 @@ impl opentelemetry_http::HttpClient for AwcOtlpHttpClient { } } -#[cfg(test)] -mod http_transport_tests { - use super::*; - use crate::webserver::http_client::transport_tests::{response_server, unavailable_url}; - use std::time::Duration; - - #[actix_web::test] - async fn preserves_methods_bodies_status_and_raw_request_headers() -> anyhow::Result<()> { - 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 request = opentelemetry_http::Request::builder() - .method(method) - .uri(url) - .header("x-request", b"\xff".as_slice()) - .body(opentelemetry_http::Bytes::from_static(b"request"))?; - let response = - execute_otlp_http_request_with_awc(&awc::Client::default(), request).await?; - assert_eq!(response.status().as_u16(), 202); - assert_eq!(response.body().as_ref(), 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())); - assert!( - sent.windows(14) - .any(|bytes| bytes == b"x-request: \xff\r\n") - ); - assert!(sent.ends_with(b"\r\n\r\nrequest")); - } - Ok(()) - } - - #[actix_web::test] - async fn rejects_non_text_response_headers_before_reading_body() -> anyhow::Result<()> { - 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 request = opentelemetry_http::Request::builder() - .uri(url) - .body(opentelemetry_http::Bytes::new())?; - let error = tokio::time::timeout( - Duration::from_secs(1), - execute_otlp_http_request_with_awc(&awc::Client::default(), request), - ) - .await? - .unwrap_err(); - assert!( - error - .to_string() - .starts_with("Invalid OTLP response header value for x-invalid:") - ); - server.abort(); - Ok(()) - } - - #[actix_web::test] - async fn retains_send_and_body_error_context() -> anyhow::Result<()> { - let request = opentelemetry_http::Request::builder() - .uri(unavailable_url().await) - .body(opentelemetry_http::Bytes::new())?; - let error = execute_otlp_http_request_with_awc(&awc::Client::default(), request) - .await - .unwrap_err(); - assert!( - error - .to_string() - .starts_with("Failed to send OTLP HTTP request to ") - ); - - let (url, server) = response_server( - b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\ninvalid\r\n", - false, - ) - .await; - let request = opentelemetry_http::Request::builder() - .uri(&url) - .body(opentelemetry_http::Bytes::new())?; - let error = execute_otlp_http_request_with_awc(&awc::Client::default(), request) - .await - .unwrap_err(); - assert!(error.to_string().starts_with(&format!( - "Failed to read OTLP HTTP response body from {url}:" - ))); - server.await?; - Ok(()) - } -} - /// Initializes logging / tracing. Returns whether `OTel` was activated. pub fn init_telemetry() -> anyhow::Result { init_telemetry_with_log_layer(logfmt::LogfmtLayer::new()) diff --git a/src/webserver/http_client.rs b/src/webserver/http_client.rs index a064527c1..3c02c09e1 100644 --- a/src/webserver/http_client.rs +++ b/src/webserver/http_client.rs @@ -140,42 +140,68 @@ pub(crate) fn get_http_client_from_appdata( } #[cfg(test)] -pub(crate) mod transport_tests { +mod transport_tests { use super::*; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use openidconnect::http::{Request, Response, header::ToStrError}; + use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpListener; use tokio::task::JoinHandle; - pub(crate) async fn response_server( + #[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>> { + 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::execute_oidc_request_with_awc(client, request).await, + Otlp => crate::telemetry::execute_otlp_http_request_with_awc( + &client, + request.map(actix_web::web::Bytes::from), + ) + .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>) { 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 (mut socket, _) = listener.accept().await.unwrap(); + let (socket, _) = listener.accept().await.unwrap(); + let mut socket = BufReader::new(socket); let mut request = Vec::new(); - let (header_end, content_length) = loop { - let mut buffer = [0; 1024]; - let count = socket.read(&mut buffer).await.unwrap(); - assert!(count > 0, "request ended before its headers"); - request.extend_from_slice(&buffer[..count]); - if let Some(end) = request.windows(4).position(|bytes| bytes == b"\r\n\r\n") { - let length = String::from_utf8_lossy(&request[..end]) - .lines() - .filter_map(|line| line.split_once(':')) - .find(|(name, _)| name.eq_ignore_ascii_case("content-length")) - .map_or(0, |(_, value)| value.trim().parse::().unwrap()); - break (end + 4, length); - } - }; - while request.len() < header_end + content_length { - let mut buffer = [0; 1024]; - let count = socket.read(&mut buffer).await.unwrap(); - assert!(count > 0, "request ended before its body"); - request.extend_from_slice(&buffer[..count]); + while !request.ends_with(b"\r\n\r\n") { + assert!(socket.read_until(b'\n', &mut request).await.unwrap() > 0); } - socket.write_all(response).await.unwrap(); + 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::().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; } @@ -184,9 +210,99 @@ pub(crate) mod transport_tests { (url, task) } - pub(crate) async fn unavailable_url() -> String { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - format!("http://{}/transport", listener.local_addr().unwrap()) + #[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!(error.downcast_ref::().is_some()); + 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(); + match adapter { + Oidc => assert!( + error + .downcast_ref::() + .is_some() + ), + Otlp => assert!( + error + .to_string() + .starts_with("Invalid OTLP response header value for 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(match adapter { + Oidc => "Failed to send request: GET ", + Otlp => "Failed to send OTLP HTTP request to ", + })); + 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(); + match adapter { + Oidc => assert_eq!(error.to_string(), format!("Couldnt read from {url}")), + Otlp => assert!(error.to_string().starts_with(&format!( + "Failed to read OTLP HTTP response body from {url}:" + ))), + } + server.await?; + } + Ok(()) } #[actix_web::test] diff --git a/src/webserver/oidc.rs b/src/webserver/oidc.rs index a25dfa7d4..4bb27d980 100644 --- a/src/webserver/oidc.rs +++ b/src/webserver/oidc.rs @@ -1033,7 +1033,7 @@ impl<'c> AsyncHttpClient<'c> for AwcHttpClient<'c> { } } -async fn execute_oidc_request_with_awc( +pub(crate) async fn execute_oidc_request_with_awc( client: Client, request: openidconnect::HttpRequest, ) -> Result>, anyhow::Error> { @@ -1399,107 +1399,6 @@ fn validate_redirect_url(url: String, redirect_uri: &str) -> String { '/'.to_string() } -#[cfg(test)] -mod http_transport_tests { - use super::*; - use crate::webserver::http_client::transport_tests::{response_server, unavailable_url}; - - #[actix_web::test] - async fn preserves_methods_bodies_status_and_headers() -> anyhow::Result<()> { - 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 request = openidconnect::http::Request::builder() - .method(method) - .uri(url) - .header("x-request", "value") - .body(b"request".to_vec())?; - let response = execute_oidc_request_with_awc(Client::default(), request).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 = String::from_utf8(server.await?)?; - assert!(sent.starts_with(&format!("{method} /transport HTTP/1.1\r\n"))); - assert!(sent.contains("x-request: value\r\n")); - assert!(sent.ends_with("\r\n\r\nrequest")); - } - Ok(()) - } - - #[actix_web::test] - async fn rejects_non_text_request_and_response_headers() -> anyhow::Result<()> { - let request = openidconnect::http::Request::builder() - .uri("http://127.0.0.1:1/transport") - .header( - "x-invalid", - openidconnect::http::HeaderValue::from_bytes(b"\xff")?, - ) - .body(Vec::new())?; - let error = execute_oidc_request_with_awc(Client::default(), request) - .await - .unwrap_err(); - assert!( - error - .downcast_ref::() - .is_some() - ); - - 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 request = openidconnect::http::Request::builder() - .uri(url) - .body(Vec::new())?; - let error = tokio::time::timeout( - Duration::from_secs(1), - execute_oidc_request_with_awc(Client::default(), request), - ) - .await? - .unwrap_err(); - assert!( - error - .downcast_ref::() - .is_some() - ); - server.abort(); - Ok(()) - } - - #[actix_web::test] - async fn retains_send_and_body_error_context() -> anyhow::Result<()> { - let request = openidconnect::http::Request::builder() - .uri(unavailable_url().await) - .body(Vec::new())?; - let error = execute_oidc_request_with_awc(Client::default(), request) - .await - .unwrap_err(); - assert!( - error - .to_string() - .starts_with("Failed to send 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 request = openidconnect::http::Request::builder() - .uri(&url) - .body(Vec::new())?; - let error = execute_oidc_request_with_awc(Client::default(), request) - .await - .unwrap_err(); - assert_eq!(error.to_string(), format!("Couldnt read from {url}")); - server.await?; - Ok(()) - } -} - #[cfg(test)] mod tests { use super::*; From 78706e827484b6b10277a9847747f8a41219512f Mon Sep 17 00:00:00 2001 From: Ophir Lojkine Date: Wed, 7 Oct 2026 11:06:30 +0000 Subject: [PATCH 3/3] Unify OIDC and OTLP HTTP request conversion --- src/telemetry.rs | 60 +-------------- src/webserver/http_client.rs | 142 ++++++++++++++++++----------------- src/webserver/oidc.rs | 61 +++------------ 3 files changed, 85 insertions(+), 178 deletions(-) diff --git a/src/telemetry.rs b/src/telemetry.rs index b8a337198..0dcd663a8 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -9,7 +9,7 @@ use std::env; use std::sync::{Once, OnceLock}; -use crate::webserver::http_client::{BufferedRequestError, send_buffered_request}; +use crate::webserver::http_client::send_http_request; use anyhow::Context as _; use opentelemetry_sdk::metrics::SdkMeterProvider; use opentelemetry_sdk::trace::SdkTracerProvider; @@ -91,9 +91,9 @@ fn init_otlp_http_worker_sender() -> anyhow::Result anyhow::Result, -) -> anyhow::Result> { - 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 (response_builder, response_body) = - send_buffered_request(awc_request, request_body, None, |status, headers| { - let mut builder = opentelemetry_http::Response::builder().status(status.as_u16()); - for (header_name, header_value) in headers { - let header_value = header_value.to_str().map_err(|error| { - anyhow::anyhow!( - "Invalid OTLP response header value for {}: {error}", - header_name.as_str() - ) - })?; - builder = builder.header(header_name.as_str(), header_value); - } - Ok(builder) - }) - .await - .map_err(|error| match error { - BufferedRequestError::Send(error) => { - anyhow::anyhow!("Failed to send OTLP HTTP request to {awc_uri}: {error}") - } - BufferedRequestError::ResponseHead(error) => error, - BufferedRequestError::Body(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( diff --git a/src/webserver/http_client.rs b/src/webserver/http_client.rs index 3c02c09e1..c7fb7f959 100644 --- a/src/webserver/http_client.rs +++ b/src/webserver/http_client.rs @@ -1,43 +1,53 @@ -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; -/// The stage that failed while buffering an outbound HTTP response. -#[derive(Debug)] -pub(crate) enum BufferedRequestError { - Send(awc::error::SendRequestError), - ResponseHead(anyhow::Error), - Body(awc::error::PayloadError), -} - -/// Sends a prepared request and validates the response head before reading its body. -/// Callers retain their header policy, timeout selection, and error context. -pub(crate) async fn send_buffered_request( - request: awc::ClientRequest, - body: B, +/// 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( + client: &awc::Client, + request: Request, body_timeout: Option, - prepare_response_head: impl FnOnce( - awc::http::StatusCode, - &awc::http::header::HeaderMap, - ) -> anyhow::Result, -) -> Result<(H, actix_web::web::Bytes), BufferedRequestError> +) -> anyhow::Result> where B: actix_web::body::MessageBody + 'static, { - let response = request - .send_body(body) + 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 - .map_err(BufferedRequestError::Send)?; - let head = prepare_response_head(response.status(), response.headers()) - .map_err(BufferedRequestError::ResponseHead)?; - let mut response = match body_timeout { - Some(timeout) => response.timeout(timeout), - None => response, - }; - let body = response.body().await.map_err(BufferedRequestError::Body)?; - Ok((head, body)) + .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 { @@ -142,7 +152,7 @@ pub(crate) fn get_http_client_from_appdata( #[cfg(test)] mod transport_tests { use super::*; - use openidconnect::http::{Request, Response, header::ToStrError}; + use openidconnect::{AsyncHttpClient, http::HeaderValue}; use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpListener; use tokio::task::JoinHandle; @@ -168,13 +178,13 @@ mod transport_tests { .body(b"request".to_vec())?; let client = awc::Client::default(); match self { - Oidc => super::super::oidc::execute_oidc_request_with_awc(client, request).await, - Otlp => crate::telemetry::execute_otlp_http_request_with_awc( - &client, - request.map(actix_web::web::Bytes::from), - ) - .await - .map(|response| response.map(|body| body.to_vec())), + 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())), } } } @@ -244,7 +254,13 @@ mod transport_tests { .request("http://127.0.0.1:1/transport", "GET", b"\xff") .await .unwrap_err(); - assert!(error.downcast_ref::().is_some()); + 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", @@ -257,18 +273,11 @@ mod transport_tests { ) .await? .unwrap_err(); - match adapter { - Oidc => assert!( - error - .downcast_ref::() - .is_some() - ), - Otlp => assert!( - error - .to_string() - .starts_with("Invalid OTLP response header value for x-invalid:") - ), - } + assert!( + error + .to_string() + .starts_with("Invalid HTTP response header: x-invalid") + ); server.abort(); } Ok(()) @@ -284,22 +293,21 @@ mod transport_tests { .request(&unavailable, "GET", b"value") .await .unwrap_err(); - assert!(error.to_string().starts_with(match adapter { - Oidc => "Failed to send request: GET ", - Otlp => "Failed to send OTLP HTTP request to ", - })); + 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(); - match adapter { - Oidc => assert_eq!(error.to_string(), format!("Couldnt read from {url}")), - Otlp => assert!(error.to_string().starts_with(&format!( - "Failed to read OTLP HTTP response body from {url}:" - ))), - } + assert_eq!( + error.to_string(), + format!("Failed to read HTTP response body from {url}") + ); server.await?; } Ok(()) @@ -309,18 +317,14 @@ mod transport_tests { 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_buffered_request( - awc::Client::default().get(url), - Vec::::new(), + let error = send_http_request( + &awc::Client::default(), + Request::builder().uri(&url).body(Vec::::new()).unwrap(), Some(Duration::from_millis(20)), - |status, _| { - assert_eq!(status, awc::http::StatusCode::OK); - Ok(()) - }, ) .await .unwrap_err(); - assert!(matches!(error, BufferedRequestError::Body(_))); + assert!(error.downcast_ref::().is_some()); server.abort(); } } diff --git a/src/webserver/oidc.rs b/src/webserver/oidc.rs index 4bb27d980..91f0b0826 100644 --- a/src/webserver/oidc.rs +++ b/src/webserver/oidc.rs @@ -5,9 +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::{ - BufferedRequestError, get_http_client_from_appdata, send_buffered_request, -}; +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, }; @@ -1024,62 +1022,21 @@ impl<'c> AsyncHttpClient<'c> for AwcHttpClient<'c> { Pin> + '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) }) } } -pub(crate) async fn execute_oidc_request_with_awc( - client: Client, - request: openidconnect::HttpRequest, -) -> Result>, 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 (resp_builder, body) = send_buffered_request( - req, - body, - Some(OIDC_HTTP_BODY_TIMEOUT), - |status, headers| { - log::debug!( - "Received OIDC response headers: status={}, content_type={:?}", - status, - headers - .get(header::CONTENT_TYPE) - .and_then(|value| value.to_str().ok()) - ); - let mut builder = openidconnect::http::Response::builder().status(status.as_u16()); - for (name, value) in headers { - builder = builder.header(name.as_str(), value.to_str()?); - } - Ok(builder) - }, - ) - .await - .map_err(|error| match error { - BufferedRequestError::Send(error) => anyhow!(error.to_string()).context(format!( - "Failed to send request: {} {}", - req_head.method, req_head.uri - )), - BufferedRequestError::ResponseHead(error) => error, - BufferedRequestError::Body(error) => { - anyhow::Error::new(error).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);