diff --git a/src/telemetry.rs b/src/telemetry.rs index 41c669edb..0dcd663a8 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::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 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 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( diff --git a/src/webserver/http_client.rs b/src/webserver/http_client.rs index 1f934b705..c7fb7f959 100644 --- a/src/webserver/http_client.rs +++ b/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( + client: &awc::Client, + request: Request, + body_timeout: Option, +) -> anyhow::Result> +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>, @@ -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>> { + 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>) { + 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::().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::::new()).unwrap(), + Some(Duration::from_millis(20)), + ) + .await + .unwrap_err(); + assert!(error.downcast_ref::().is_some()); + server.abort(); + } +} diff --git a/src/webserver/oidc.rs b/src/webserver/oidc.rs index 153284894..91f0b0826 100644 --- a/src/webserver/oidc.rs +++ b/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> + '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>, 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);