From 5a8715b9a88fa430ab707a59c173d31c3a7088f9 Mon Sep 17 00:00:00 2001 From: jamie Date: Sun, 27 Sep 2026 16:56:48 +0100 Subject: [PATCH] test: cover cut-off request bodies and http/2 clients --- tests/integration/body_failure.rs | 221 ++++++++++++++++++++++++++ tests/integration/http2.rs | 251 ++++++++++++++++++++++++++++++ tests/integration/main.rs | 2 + tests/integration/tls.rs | 2 +- 4 files changed, 475 insertions(+), 1 deletion(-) create mode 100644 tests/integration/body_failure.rs create mode 100644 tests/integration/http2.rs diff --git a/tests/integration/body_failure.rs b/tests/integration/body_failure.rs new file mode 100644 index 0000000..611f4f6 --- /dev/null +++ b/tests/integration/body_failure.rs @@ -0,0 +1,221 @@ +use crate::common; + +use bytes::Bytes; +use http_body_util::Empty; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::oneshot; +use tokio::time::timeout; + +/// longer than any step in these tests should take on a loaded CI runner. +const PATIENCE: Duration = Duration::from_secs(10); + +/// what a raw origin read on its one connection. +struct Received { + bytes: Vec, + /// the bytes held a whole request, which the origin then answered + complete: bool, + /// the proxy closed the connection + closed: bool, +} + +impl std::fmt::Debug for Received { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Received") + .field("bytes", &String::from_utf8_lossy(&self.bytes)) + .field("complete", &self.complete) + .field("closed", &self.closed) + .finish() + } +} + +impl Received { + fn body(&self) -> &[u8] { + let end = find(&self.bytes, b"\r\n\r\n").expect("no end of request head"); + &self.bytes[end + 4..] + } + + fn head(&self) -> String { + let end = find(&self.bytes, b"\r\n\r\n").expect("no end of request head"); + String::from_utf8_lossy(&self.bytes[..end]).to_ascii_lowercase() + } +} + +fn find(haystack: &[u8], needle: &[u8]) -> Option { + haystack.windows(needle.len()).position(|w| w == needle) +} + +/// whether `bytes` hold a whole request with a `Content-Length` or chunked +/// body. the test bodies hold only letters, so a `0\r\n\r\n` can only be the +/// last chunk. +fn is_complete(bytes: &[u8]) -> bool { + let Some(end) = find(bytes, b"\r\n\r\n") else { + return false; + }; + let head = String::from_utf8_lossy(&bytes[..end]).to_ascii_lowercase(); + let body = &bytes[end + 4..]; + let length = head + .lines() + .find_map(|line| line.strip_prefix("content-length:")) + .map(|value| value.trim().parse::().unwrap()); + match length { + Some(length) => body.len() >= length, + None => body.starts_with(b"0\r\n\r\n") || find(body, b"\r\n0\r\n\r\n").is_some(), + } +} + +/// an origin that accepts one connection and records what arrives on it. it +/// answers 200 only once it holds a whole request. `body_started` fires +/// once the request head and some body bytes have arrived. +async fn raw_origin() -> (String, oneshot::Receiver<()>, oneshot::Receiver) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let (started_tx, started_rx) = oneshot::channel(); + let (received_tx, received_rx) = oneshot::channel(); + actix_web::rt::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let received = record(stream, started_tx).await; + let _ = received_tx.send(received); + }); + (url, started_rx, received_rx) +} + +async fn record(mut stream: TcpStream, started: oneshot::Sender<()>) -> Received { + let mut started = Some(started); + let mut bytes = Vec::new(); + let mut buf = [0; 4096]; + loop { + let n = match timeout(PATIENCE, stream.read(&mut buf)).await { + Ok(Ok(n)) => n, + // a reset closes the connection as surely as an EOF + Ok(Err(_)) => 0, + Err(_) => { + return Received { + bytes, + complete: false, + closed: false, + } + } + }; + if n == 0 { + return Received { + bytes, + complete: false, + closed: true, + }; + } + bytes.extend_from_slice(&buf[..n]); + if find(&bytes, b"\r\n\r\n").is_some_and(|end| bytes.len() > end + 4) { + if let Some(started) = started.take() { + let _ = started.send(()); + } + } + if is_complete(&bytes) { + let ok = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + stream.write_all(ok).await.unwrap(); + return Received { + bytes, + complete: true, + closed: false, + }; + } + } +} + +/// sends `head` and `start` to the proxy, waits until the origin holds part +/// of the body, then sends `rest` and half-closes the connection. returns +/// what the origin received and the proxy's whole response. +async fn interrupted_request(head: &str, start: &[u8], rest: &[u8]) -> (Received, String) { + let (origin_url, body_started, received) = raw_origin().await; + let proxy = common::spawn(&origin_url); + + let mut client = TcpStream::connect(proxy.addr).await.unwrap(); + client.write_all(head.as_bytes()).await.unwrap(); + client.write_all(start).await.unwrap(); + timeout(PATIENCE, body_started) + .await + .expect("the origin received no body") + .unwrap(); + client.write_all(rest).await.unwrap(); + client.shutdown().await.unwrap(); + + let received = timeout(PATIENCE, received) + .await + .expect("the origin connection stayed open") + .unwrap(); + let mut response = Vec::new(); + timeout(PATIENCE, client.read_to_end(&mut response)) + .await + .expect("the proxy kept the client connection open") + .unwrap(); + + assert_healthy(&proxy).await; + proxy.stop().await; + (received, String::from_utf8_lossy(&response).into_owned()) +} + +async fn assert_healthy(proxy: &common::Running) { + let client = common::client::>(); + let resp = timeout(PATIENCE, client.get(proxy.url("/health").parse().unwrap())) + .await + .expect("health check timed out") + .unwrap(); + assert_eq!(resp.status(), 200); +} + +#[actix_web::test] +async fn content_length_body_cut_short_does_not_reach_origin_whole() { + let head = "POST /submit HTTP/1.1\r\nHost: localhost\r\nContent-Length: 100\r\n\r\n"; + + let (received, response) = interrupted_request(head, &[b'a'; 40], b"").await; + + assert!( + !received.complete, + "origin got a whole request: {received:?}" + ); + assert!(received.closed); + assert!(received.head().contains("content-length: 100")); + assert!(received.body().len() < 100, "{received:?}"); + assert!( + response.starts_with("HTTP/1.1 502 Bad Gateway\r\n"), + "unexpected response: {response}" + ); +} + +#[actix_web::test] +async fn chunked_body_cut_short_gets_no_last_chunk_at_origin() { + let head = "POST /submit HTTP/1.1\r\nHost: localhost\r\nTransfer-Encoding: chunked\r\n\r\n"; + + let (received, response) = interrupted_request(head, b"5\r\nhello\r\n", b"").await; + + assert!( + !received.complete, + "origin got a whole request: {received:?}" + ); + assert!(received.closed); + assert!(received.head().contains("transfer-encoding: chunked")); + assert!(find(received.body(), b"hello").is_some(), "{received:?}"); + assert!( + response.starts_with("HTTP/1.1 502 Bad Gateway\r\n"), + "unexpected response: {response}" + ); +} + +#[actix_web::test] +async fn malformed_chunk_gets_no_last_chunk_at_origin() { + let head = "POST /submit HTTP/1.1\r\nHost: localhost\r\nTransfer-Encoding: chunked\r\n\r\n"; + + let (received, response) = + interrupted_request(head, b"5\r\nhello\r\n", b"zz\r\nworld\r\n0\r\n\r\n").await; + + assert!( + !received.complete, + "origin got a whole request: {received:?}" + ); + assert!(received.closed); + assert!(find(received.body(), b"hello").is_some(), "{received:?}"); + assert!(find(received.body(), b"world").is_none(), "{received:?}"); + // actix closes the connection on a framing error while the handler runs + assert!(response.is_empty(), "unexpected response: {response}"); +} diff --git a/tests/integration/http2.rs b/tests/integration/http2.rs new file mode 100644 index 0000000..34c4f36 --- /dev/null +++ b/tests/integration/http2.rs @@ -0,0 +1,251 @@ +use crate::tls::spawn_tls; + +use bytes::Bytes; +use http_body_util::{BodyExt, Empty, Full}; +use hyper::{Request, Response, Version}; +use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder}; +use hyper_util::client::legacy::connect::HttpConnector; +use hyper_util::client::legacy::Client; +use hyper_util::rt::TokioExecutor; +use std::net::SocketAddr; +use tokio_rustls::rustls::pki_types::CertificateDer; +use tokio_rustls::rustls::{ClientConfig, RootCertStore}; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +type H2Client = Client, B>; + +/// an HTTP/2-only client that trusts only `cert_der` and offers only `h2` +/// through ALPN, so a handshake that falls back to HTTP/1.1 fails. +fn h2_client(cert_der: Vec) -> H2Client +where + B: hyper::body::Body + Send + Unpin + 'static, + B::Data: Send, + B::Error: Into>, +{ + let mut roots = RootCertStore::empty(); + roots.add(CertificateDer::from(cert_der)).unwrap(); + let tls = ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(); + let https = HttpsConnectorBuilder::new() + .with_tls_config(tls) + .https_only() + .enable_http2() + .build(); + Client::builder(TokioExecutor::new()) + .http2_only(true) + .build(https) +} + +fn url(addr: SocketAddr, path: &str) -> String { + format!("https://localhost:{}{path}", addr.port()) +} + +async fn get(client: &H2Client>, addr: SocketAddr, path: &str) -> Response { + let resp = client + .request(Request::get(url(addr, path)).body(Empty::new()).unwrap()) + .await + .expect("h2 request"); + let (parts, body) = resp.into_parts(); + let body = body.collect().await.unwrap().to_bytes(); + Response::from_parts(parts, body) +} + +fn header<'a>(resp: &'a Response, name: &str) -> &'a str { + resp.headers() + .get(name) + .unwrap_or_else(|| panic!("no {name} header")) + .to_str() + .unwrap() +} + +#[actix_web::test] +async fn h2_get_is_proxied() { + let origin = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/page")) + .respond_with( + ResponseTemplate::new(201) + .insert_header("x-origin", "test") + .set_body_string("hello over h2"), + ) + .expect(1) + .mount(&origin) + .await; + let ((https_addr, _), handle, cert_der, _dirs) = spawn_tls(&origin.uri()); + let client = h2_client(cert_der); + + let resp = get(&client, https_addr, "/page").await; + + assert_eq!(resp.version(), Version::HTTP_2); + assert_eq!(resp.status(), 201); + assert_eq!(header(&resp, "x-origin"), "test"); + assert_eq!(resp.body().as_ref(), b"hello over h2"); + + drop(client); + handle.stop(true).await; +} + +#[actix_web::test] +async fn h2_post_body_reaches_origin_intact() { + let origin = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/submit")) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&origin) + .await; + let ((https_addr, _), handle, cert_der, _dirs) = spawn_tls(&origin.uri()); + let client = h2_client(cert_der); + // larger than the 64 KiB initial h2 flow-control window, and not one + // repeated byte, so a truncated or reordered copy does not compare equal + let body: Vec = (0..200_000).map(|i| (i % 251) as u8).collect(); + + let resp = client + .request( + Request::post(url(https_addr, "/submit")) + .body(Full::new(Bytes::from(body.clone()))) + .unwrap(), + ) + .await + .expect("h2 request"); + + assert_eq!(resp.version(), Version::HTTP_2); + assert_eq!(resp.status(), 200); + drop(resp); + let received = origin.received_requests().await.unwrap(); + assert_eq!(received.len(), 1); + assert_eq!(received[0].body.len(), body.len()); + assert!(received[0].body == body, "the origin body differs"); + + drop(client); + handle.stop(true).await; +} + +#[actix_web::test] +async fn h2_cacheable_response_is_a_miss_then_a_hit() { + let origin = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/cached")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=60") + .set_body_string("cached body"), + ) + .expect(1) + .mount(&origin) + .await; + let ((https_addr, _), handle, cert_der, _dirs) = spawn_tls(&origin.uri()); + let client = h2_client(cert_der); + + let first = get(&client, https_addr, "/cached").await; + let second = get(&client, https_addr, "/cached").await; + + for resp in [&first, &second] { + assert_eq!(resp.version(), Version::HTTP_2); + assert_eq!(resp.status(), 200); + assert_eq!(resp.body().as_ref(), b"cached body"); + } + assert_eq!(header(&first, "x-shadowstep-cache"), "MISS"); + assert_eq!(header(&second, "x-shadowstep-cache"), "HIT"); + assert_eq!(origin.received_requests().await.unwrap().len(), 1); + + drop(client); + handle.stop(true).await; +} + +#[actix_web::test] +async fn h2_request_reaches_origin_as_https_with_the_authority_as_host() { + let origin = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/page")) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&origin) + .await; + let ((https_addr, _), handle, cert_der, _dirs) = spawn_tls(&origin.uri()); + let client = h2_client(cert_der); + + // an HTTP/2 request carries its host only in `:authority` + let resp = get(&client, https_addr, "/page").await; + + assert_eq!(resp.version(), Version::HTTP_2); + assert_eq!(resp.status(), 200); + let received = origin.received_requests().await.unwrap(); + assert_eq!(received.len(), 1); + let values = |name: &str| -> Vec { + received[0] + .headers + .get_all(name) + .iter() + .map(|v| v.to_str().unwrap().to_owned()) + .collect() + }; + assert_eq!(values("x-forwarded-proto"), ["https"]); + assert_eq!( + values("x-forwarded-host"), + [format!("localhost:{}", https_addr.port())] + ); + + drop(client); + handle.stop(true).await; +} + +/// the h2 crate refuses to send a response that holds a connection-specific +/// field (RFC 9113 section 8.2.2), and a client treats one it receives as +/// malformed, so the proxy must remove them from both fresh and stored +/// responses. actix removes most of them itself, but not `TE` or the fields +/// that `Connection` lists. +#[actix_web::test] +async fn h2_responses_carry_no_connection_specific_fields() { + let origin = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/page")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=60") + .insert_header("connection", "x-hop") + .insert_header("x-hop", "1") + .insert_header("keep-alive", "timeout=5") + .insert_header("proxy-connection", "keep-alive") + .insert_header("te", "gzip") + .insert_header("upgrade", "h2c") + .insert_header("x-end-to-end", "kept") + .set_body_string("body"), + ) + .expect(1) + .mount(&origin) + .await; + let ((https_addr, _), handle, cert_der, _dirs) = spawn_tls(&origin.uri()); + let client = h2_client(cert_der); + + let fresh = get(&client, https_addr, "/page").await; + let stored = get(&client, https_addr, "/page").await; + + assert_eq!(header(&fresh, "x-shadowstep-cache"), "MISS"); + assert_eq!(header(&stored, "x-shadowstep-cache"), "HIT"); + for resp in [&fresh, &stored] { + assert_eq!(resp.version(), Version::HTTP_2); + assert_eq!(resp.status(), 200); + for name in [ + "connection", + "x-hop", + "keep-alive", + "proxy-connection", + "te", + "transfer-encoding", + "upgrade", + ] { + assert!( + resp.headers().get(name).is_none(), + "{name} reached the client" + ); + } + assert_eq!(header(resp, "x-end-to-end"), "kept"); + assert_eq!(resp.body().as_ref(), b"body"); + } + + drop(client); + handle.stop(true).await; +} diff --git a/tests/integration/main.rs b/tests/integration/main.rs index 08b4dbf..fd848c2 100644 --- a/tests/integration/main.rs +++ b/tests/integration/main.rs @@ -1,6 +1,8 @@ +mod body_failure; mod cache; mod common; mod forwarded; +mod http2; mod proxy; mod smoke; mod tls; diff --git a/tests/integration/tls.rs b/tests/integration/tls.rs index 8991b4b..8ae59fd 100644 --- a/tests/integration/tls.rs +++ b/tests/integration/tls.rs @@ -31,7 +31,7 @@ fn self_signed_cert() -> (TempDir, Vec) { /// a proxy for `origin_url` listening on HTTP and HTTPS ports, both /// ephemeral. returns the HTTPS and HTTP addresses, the server handle, the /// certificate DER for the client to trust, and the directories to keep. -fn spawn_tls( +pub(crate) fn spawn_tls( origin_url: &str, ) -> ( (SocketAddr, SocketAddr),