From cb50602981a7235f29e3e363e72e6be040c4a2c3 Mon Sep 17 00:00:00 2001 From: sub2api-patch automation Date: Mon, 19 Oct 2026 20:32:12 +0000 Subject: [PATCH 1/2] feat: support bounded large Responses HTTP requests --- src/request.rs | 14 +- src/server.rs | 97 ++++++----- tests/cases/responses_limits.rs | 286 ++++++++++++++++++++++++++++++++ tests/gateway.rs | 11 +- 4 files changed, 367 insertions(+), 41 deletions(-) create mode 100644 tests/cases/responses_limits.rs diff --git a/src/request.rs b/src/request.rs index 8e97a88..e508c6a 100644 --- a/src/request.rs +++ b/src/request.rs @@ -28,6 +28,18 @@ pub struct CreateResponse { pub service_tier: Option, pub prompt_cache_key: Option, pub include: Option>, + #[serde(default, deserialize_with = "deserialize_client_metadata")] + pub client_metadata: Option>, +} + +// Compact rejects this field even when empty; null must not act like omission. +fn deserialize_client_metadata<'de, D>( + deserializer: D, +) -> Result>, D::Error> +where + D: serde::Deserializer<'de>, +{ + HashMap::deserialize(deserializer).map(Some) } #[derive(Debug, Deserialize)] @@ -266,7 +278,7 @@ impl CreateResponse { .unwrap_or_else(|| session_id.to_owned()), ), text, - client_metadata: Some(HashMap::new()), + client_metadata: Some(self.client_metadata.unwrap_or_default()), access_programs: None, }) } diff --git a/src/server.rs b/src/server.rs index 76e68e8..fbd3e8c 100644 --- a/src/server.rs +++ b/src/server.rs @@ -2,7 +2,7 @@ use crate::{CODEX_REV, backend::Backend, error::GatewayError, request::CreateRes use axum::{ Json, Router, body::Body, - extract::{DefaultBodyLimit, Query, State, rejection::JsonRejection}, + extract::{DefaultBodyLimit, FromRequest, Query, Request, State, rejection::JsonRejection}, http::{HeaderMap, StatusCode}, response::{IntoResponse, Response}, routing::{get, post}, @@ -13,6 +13,8 @@ use serde_json::{Value, json}; use std::{sync::Arc, time::Duration}; use tokio::sync::Semaphore; +const RESPONSES_BODY_LIMIT: usize = 64 * 1024 * 1024; + pub struct Gateway { pub backend: Backend, pub models: Vec, @@ -41,15 +43,24 @@ pub fn router(gateway: Gateway) -> Router { .route("/codex/images/edits", post(crate::images::edit)) .route("/backend-api/codex/images/generations", post(crate::images::generate)) .route("/backend-api/codex/images/edits", post(crate::images::edit)) - .route("/v1/responses", post(create).get(crate::websocket::upgrade)) + .route( + "/v1/responses", + post(create) + .layer(DefaultBodyLimit::max(RESPONSES_BODY_LIMIT)) + .get(crate::websocket::upgrade), + ) .route( "/codex/responses", - post(native_create).get(crate::websocket::upgrade), + post(native_create) + .layer(DefaultBodyLimit::max(RESPONSES_BODY_LIMIT)) + .get(crate::websocket::upgrade), ) .route("/codex/models", get(native_models)) .route( "/backend-api/codex/responses", - post(native_create).get(crate::websocket::upgrade), + post(native_create) + .layer(DefaultBodyLimit::max(RESPONSES_BODY_LIMIT)) + .get(crate::websocket::upgrade), ) .route("/backend-api/codex/models", get(native_models)) .route("/v1/responses/compact", post(compact)) @@ -124,16 +135,19 @@ async fn models( } async fn create( State(gateway): State>, - headers: HeaderMap, - body: Result, JsonRejection>, + request: Request, ) -> Result { + let headers = request.headers().clone(); authorize(&gateway, &headers)?; - let Json(input) = body.map_err(|e| GatewayError { - upstream_response: None, - status: e.status(), - code: "invalid_request_error", - message: e.body_text(), - })?; + let (parts, body) = request.into_parts(); + let body = Bytes::from_request(Request::from_parts(parts.clone(), body), &()) + .await + .map_err(|e| crate::standalone::extraction_error(e.status(), e.body_text()))?; + let body = decode_responses_body(&headers, body).await?; + let Json(input) = + Json::::from_request(Request::from_parts(parts, Body::from(body)), &()) + .await + .map_err(|e| crate::standalone::extraction_error(e.status(), e.body_text()))?; let model = gateway .models .iter() @@ -222,52 +236,54 @@ async fn native_create( body: Bytes, ) -> Result { authorize(&gateway, &headers)?; - let body = match headers + let body = decode_responses_body(&headers, body).await?; + let raw: Box = + serde_json::from_slice(&body).map_err(|e| GatewayError::invalid(e.to_string()))?; + let body: Value = + serde_json::from_str(raw.get()).map_err(|e| GatewayError::invalid(e.to_string()))?; + if !body.is_object() || body["stream"] != true || !body["model"].is_string() { + return Err(GatewayError::invalid( + "native endpoint requires an object with model and stream: true", + )); + } + let forwarded = forward_request_headers(&headers); + respond(gateway, None, Some((raw, forwarded)), true, false).await +} + +async fn decode_responses_body(headers: &HeaderMap, body: Bytes) -> Result { + match headers .get("content-encoding") .and_then(|v| v.to_str().ok()) { - None | Some("identity") => body.to_vec(), + None | Some("identity") => Ok(body), Some("zstd") => tokio::task::spawn_blocking(move || { use std::io::Read; let decoder = zstd::stream::read::Decoder::new(body.as_ref()) .map_err(|_| GatewayError::invalid("invalid zstd request"))?; let mut decoded = Vec::new(); decoder - .take(16 * 1024 * 1024 + 1) + .take(RESPONSES_BODY_LIMIT as u64 + 1) .read_to_end(&mut decoded) .map_err(|_| GatewayError::invalid("invalid zstd request"))?; - if decoded.len() > 16 * 1024 * 1024 { + if decoded.len() > RESPONSES_BODY_LIMIT { return Err(GatewayError { upstream_response: None, status: StatusCode::PAYLOAD_TOO_LARGE, code: "invalid_request_error", - message: "decoded body exceeds 16 MiB".into(), + message: "decoded body exceeds 64 MiB".into(), }); } - Ok(decoded) + Ok(Bytes::from(decoded)) }) .await - .map_err(GatewayError::internal)??, - _ => { - return Err(GatewayError { - upstream_response: None, - status: StatusCode::UNSUPPORTED_MEDIA_TYPE, - code: "invalid_request_error", - message: "supported content encodings: identity, zstd".into(), - }); - } - }; - let raw: Box = - serde_json::from_slice(&body).map_err(|e| GatewayError::invalid(e.to_string()))?; - let body: Value = - serde_json::from_str(raw.get()).map_err(|e| GatewayError::invalid(e.to_string()))?; - if !body.is_object() || body["stream"] != true || !body["model"].is_string() { - return Err(GatewayError::invalid( - "native endpoint requires an object with model and stream: true", - )); + .map_err(GatewayError::internal)?, + _ => Err(GatewayError { + upstream_response: None, + status: StatusCode::UNSUPPORTED_MEDIA_TYPE, + code: "invalid_request_error", + message: "supported content encodings: identity, zstd".into(), + }), } - let forwarded = forward_request_headers(&headers); - respond(gateway, None, Some((raw, forwarded)), true, false).await } pub(crate) fn forward_request_headers(headers: &HeaderMap) -> HeaderMap { @@ -451,6 +467,11 @@ async fn compact( authorize(&gateway, &headers)?; let headers = crate::headers::request_headers(&headers); let Json(input) = body.map_err(|e| GatewayError::invalid(e.body_text()))?; + if input.client_metadata.is_some() { + return Err(GatewayError::invalid( + "client_metadata is not supported for compact", + )); + } let model = gateway .models .iter() diff --git a/tests/cases/responses_limits.rs b/tests/cases/responses_limits.rs new file mode 100644 index 0000000..d17437d --- /dev/null +++ b/tests/cases/responses_limits.rs @@ -0,0 +1,286 @@ +use super::*; + +const HTTP_LIMIT: usize = 64 * 1024 * 1024; + +#[tokio::test] +async fn native_http_preserves_done_items_when_terminal_output_is_empty() { + let mut fixture = events(); + *fixture.last_mut().unwrap() = completed(vec![]); + let h = Harness::new( + fixture.clone(), + StatusCode::OK, + false, + Duration::from_secs(5), + ) + .await; + let response = reqwest::Client::new() + .post(format!("{}/backend-api/codex/responses", h.url)) + .bearer_auth("client-key") + .json(&json!({"model":h.model,"stream":true,"input":[]})) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(parse_sse(&response.text().await.unwrap()), fixture); +} + +#[tokio::test] +async fn responses_accept_large_images_over_http_with_identity_and_zstd() { + for path in [ + "/v1/responses", + "/codex/responses", + "/backend-api/codex/responses", + ] { + for compressed in [false, true] { + let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(30)).await; + let image = format!("data:image/png;base64,{}", "A".repeat(17 * 1024 * 1024)); + let body = json!({"model":h.model,"stream":true,"input":[{ + "type":"message","role":"user","content":[{"type":"input_image","image_url":image}] + }]}); + let wire = serde_json::to_vec(&body).unwrap(); + let mut request = reqwest::Client::new() + .post(format!("{}{path}", h.url)) + .bearer_auth("client-key") + .header("content-type", "application/json"); + let wire = if compressed { + request = request.header("content-encoding", "zstd"); + zstd::encode_all(wire.as_slice(), 1).unwrap() + } else { + wire + }; + let response = request.body(wire).send().await.unwrap(); + assert_eq!( + response.status(), + StatusCode::OK, + "{path}, compressed={compressed}" + ); + assert!( + response + .text() + .await + .unwrap() + .contains("response.completed") + ); + let received = h.fake.received.lock().unwrap(); + assert_eq!(received.len(), 1); + assert!(received[0].1["input"][0]["content"][0]["image_url"] == image); + } + } +} + +#[tokio::test] +async fn responses_http_enforces_wire_and_decoded_boundaries() { + for path in [ + "/v1/responses", + "/codex/responses", + "/backend-api/codex/responses", + ] { + for compressed in [false, true] { + let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(30)).await; + for size in [HTTP_LIMIT, HTTP_LIMIT + 1] { + let mut wire = serde_json::to_vec(&json!({"model":h.model,"stream":true,"input":[{ + "type":"message","role":"user","content":[{"type":"input_text","text":"hello"}] + }]})) + .unwrap(); + wire.resize(size, b' '); + let mut request = reqwest::Client::new() + .post(format!("{}{path}", h.url)) + .bearer_auth("client-key") + .header("content-type", "application/json"); + let wire = if compressed { + request = request.header("content-encoding", "zstd"); + zstd::encode_all(wire.as_slice(), 1).unwrap() + } else { + wire + }; + let response = request.body(wire).send().await.unwrap(); + let expected = if size == HTTP_LIMIT { + StatusCode::OK + } else { + StatusCode::PAYLOAD_TOO_LARGE + }; + assert_eq!( + response.status(), + expected, + "{path}, compressed={compressed}, size={size}" + ); + let text = response.text().await.unwrap(); + if size == HTTP_LIMIT { + assert!(text.contains("response.completed")); + } + } + assert_eq!(h.fake.received.lock().unwrap().len(), 1); + } + } +} + +#[tokio::test] +async fn responses_preserves_public_metadata_without_accepting_continuation_or_unknown_fields() { + let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(5)).await; + let metadata = json!({"trace":"keep","ws_request_header_x_codex_turn_metadata":"turn"}); + let response = h + .request(json!({"input":"hello","client_metadata":metadata})) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response.json::().await.unwrap()["status"], + "completed" + ); + assert_eq!( + h.fake.received.lock().unwrap()[0].1["client_metadata"], + metadata + ); + for field in [ + json!({"client_metadata":{"trace":1}}), + json!({"client_metadata":null}), + json!({"previous_response_id":"resp_1"}), + json!({"stream_id":"lane"}), + json!({"unknown_field":true}), + ] { + let mut body = field; + body["input"] = json!("hello"); + let response = h.request(body).send().await.unwrap(); + assert!(response.status().is_client_error()); + } + assert_eq!(h.fake.received.lock().unwrap().len(), 1); +} + +#[tokio::test] +async fn responses_compression_validates_encoding_and_public_content_type() { + let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(5)).await; + for path in [ + "/v1/responses", + "/codex/responses", + "/backend-api/codex/responses", + ] { + for (encoding, status) in [ + ("gzip", StatusCode::UNSUPPORTED_MEDIA_TYPE), + ("zstd", StatusCode::BAD_REQUEST), + ] { + let response = reqwest::Client::new() + .post(format!("{}{path}", h.url)) + .bearer_auth("client-key") + .header("content-type", "application/json") + .header("content-encoding", encoding) + .body("invalid compressed body") + .send() + .await + .unwrap(); + assert_eq!(response.status(), status); + } + } + let response = reqwest::Client::new() + .post(format!("{}/v1/responses", h.url)) + .bearer_auth("client-key") + .body(json!({"model":h.model,"input":"hello"}).to_string()) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE); + assert!(h.fake.received.lock().unwrap().is_empty()); +} + +#[tokio::test] +async fn responses_rejects_truncated_zstd_and_oversized_compressed_wire() { + use std::io::Write; + + let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(5)).await; + let body = serde_json::to_vec(&json!({"model":h.model,"stream":true,"input":"hello"})).unwrap(); + let mut encoder = zstd::Encoder::new(Vec::new(), 1).unwrap(); + encoder.include_checksum(true).unwrap(); + encoder.write_all(&body).unwrap(); + let wire = encoder.finish().unwrap(); + let mut truncated = wire.clone(); + truncated.pop(); + // Skippable padding grows only wire bytes, isolating the compressed ingress cap. + let mut oversized = wire; + oversized.extend_from_slice(&0x184D_2A50u32.to_le_bytes()); + let padding_len = HTTP_LIMIT + 1 - oversized.len() - 4; + oversized.extend_from_slice(&(padding_len as u32).to_le_bytes()); + oversized.resize(HTTP_LIMIT + 1, 0); + assert_eq!(zstd::decode_all(oversized.as_slice()).unwrap(), body); + for path in [ + "/v1/responses", + "/codex/responses", + "/backend-api/codex/responses", + ] { + for (wire, status) in [ + (&truncated, StatusCode::BAD_REQUEST), + (&oversized, StatusCode::PAYLOAD_TOO_LARGE), + ] { + let response = reqwest::Client::new() + .post(format!("{}{path}", h.url)) + .bearer_auth("client-key") + .header("content-type", "application/json") + .header("content-encoding", "zstd") + .body(wire.clone()) + .send() + .await + .unwrap(); + assert_eq!(response.status(), status, "{path}"); + } + } + assert!(h.fake.received.lock().unwrap().is_empty()); +} + +#[tokio::test] +async fn compact_rejects_response_metadata_including_null() { + let h = Harness::new( + vec![completed(vec![ + json!({"type":"compaction","id":"cmp_1","encrypted_content":"fixture"}), + ])], + StatusCode::OK, + false, + Duration::from_secs(5), + ) + .await; + for path in ["/v1/responses/compact", "/codex/responses/compact"] { + for metadata in [json!({"trace":"keep"}), json!({}), Value::Null] { + let response = reqwest::Client::new() + .post(format!("{}{path}", h.url)) + .bearer_auth("client-key") + .json(&json!({"model":h.model,"input":"hello","client_metadata":metadata})) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST, "{path}"); + } + let duplicated = format!( + r#"{{"model":"{}","input":"hello","input":"duplicate"}}"#, + h.model + ); + let response = reqwest::Client::new() + .post(format!("{}{path}", h.url)) + .bearer_auth("client-key") + .header("content-type", "application/json") + .body(duplicated) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + assert!(h.fake.received.lock().unwrap().is_empty()); +} + +#[tokio::test] +async fn responses_larger_budget_does_not_expand_unrelated_endpoints() { + let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(5)).await; + for (path, status) in [ + ("/v1/responses/compact", StatusCode::BAD_REQUEST), + ("/v1/images/generations", StatusCode::PAYLOAD_TOO_LARGE), + ] { + let response = reqwest::Client::new() + .post(format!("{}{path}", h.url)) + .bearer_auth("client-key") + .header("content-type", "application/json") + .body(vec![b' '; 16 * 1024 * 1024 + 1]) + .send() + .await + .unwrap(); + assert_eq!(response.status(), status, "{path}"); + assert!(response.text().await.unwrap().contains("length limit")); + } + assert!(h.fake.received.lock().unwrap().is_empty()); +} diff --git a/tests/gateway.rs b/tests/gateway.rs index e362257..fdaaa24 100644 --- a/tests/gateway.rs +++ b/tests/gateway.rs @@ -2,6 +2,8 @@ mod file_cases; #[path = "cases/native.rs"] mod native_cases; +#[path = "cases/responses_limits.rs"] +mod responses_limit_cases; mod support; use axum::{ Router, @@ -220,7 +222,12 @@ impl Harness { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let upstream_url = format!("http://{}", listener.local_addr().unwrap()); let app = Router::new() - .route("/responses", post(upstream).get(upstream_ws)) + .route( + "/responses", + post(upstream) + .get(upstream_ws) + .layer(axum::extract::DefaultBodyLimit::max(65 * 1024 * 1024)), + ) .route("/realtime/calls", post(realtime_cases::upstream_call)) .route("/realtime", axum::routing::get(upstream_ws)) .route( @@ -683,7 +690,7 @@ async fn native_rejects_unauthorized_and_nonstream_and_decompression_bombs() { .status(), StatusCode::BAD_REQUEST ); - let bomb = zstd::encode_all(vec![b' '; 16 * 1024 * 1024 + 1].as_slice(), 1).unwrap(); + let bomb = zstd::encode_all(vec![b' '; 64 * 1024 * 1024 + 1].as_slice(), 1).unwrap(); assert_eq!( client .post(&url) From b0543e48d53185776fee40d3ae97064f559972ca Mon Sep 17 00:00:00 2001 From: sub2api-patch automation Date: Mon, 19 Oct 2026 23:31:24 +0000 Subject: [PATCH 2/2] Reuse native decoding and existing fixtures for Responses limits --- src/request.rs | 14 +- src/server.rs | 64 +++---- tests/cases/responses_limits.rs | 322 -------------------------------- tests/gateway.rs | 31 +-- 4 files changed, 49 insertions(+), 382 deletions(-) delete mode 100644 tests/cases/responses_limits.rs diff --git a/src/request.rs b/src/request.rs index 21cb64d..a9d2a7d 100644 --- a/src/request.rs +++ b/src/request.rs @@ -32,18 +32,6 @@ pub struct CreateResponse { /// Compatibility hint only: the Codex upstream has no output-token limit. pub max_output_tokens: Option, pub include: Option>, - #[serde(default, deserialize_with = "deserialize_client_metadata")] - pub client_metadata: Option>, -} - -// Compact rejects this field even when empty; null must not act like omission. -fn deserialize_client_metadata<'de, D>( - deserializer: D, -) -> Result>, D::Error> -where - D: serde::Deserializer<'de>, -{ - HashMap::deserialize(deserializer).map(Some) } #[derive(Debug, Deserialize)] @@ -304,7 +292,7 @@ impl CreateResponse { .unwrap_or_else(|| session_id.to_owned()), ), text, - client_metadata: Some(self.client_metadata.unwrap_or_default()), + client_metadata: Some(HashMap::new()), access_programs: None, }) } diff --git a/src/server.rs b/src/server.rs index a207476..594225d 100644 --- a/src/server.rs +++ b/src/server.rs @@ -165,7 +165,6 @@ async fn compatible_create( if compressed || native { return native_create(State(gateway), parts.headers, bytes).await; } - let bytes = decode_responses_body(&parts.headers, bytes).await?; let headers = parts.headers.clone(); let input = Json::::from_request( axum::extract::Request::from_parts(parts, Body::from(bytes)), @@ -286,31 +285,11 @@ async fn native_create( body: Bytes, ) -> Result { authorize(&gateway, &headers)?; - let body = decode_responses_body(&headers, body).await?; - let raw: Box = - serde_json::from_slice(&body).map_err(|e| GatewayError::invalid(e.to_string()))?; - let body: Value = - serde_json::from_str(raw.get()).map_err(|e| GatewayError::invalid(e.to_string()))?; - if !body.is_object() || body["stream"] != true || !body["model"].is_string() { - return Err(GatewayError::invalid( - "native endpoint requires an object with model and stream: true", - )); - } - let forwarded = forward_request_headers(&headers); - let usage = crate::telemetry::RequestTelemetry::new(gateway.telemetry.clone(), &headers).start( - body["model"].as_str().unwrap_or("unknown"), - "responses", - "http", - ); - respond(gateway, None, Some((raw, forwarded)), true, false, usage).await -} - -async fn decode_responses_body(headers: &HeaderMap, body: Bytes) -> Result { - match headers + let body = match headers .get("content-encoding") .and_then(|v| v.to_str().ok()) { - None | Some("identity") => Ok(body), + None | Some("identity") => body.to_vec(), Some("zstd") => tokio::task::spawn_blocking(move || { use std::io::Read; let decoder = zstd::stream::read::Decoder::new(body.as_ref()) @@ -328,17 +307,35 @@ async fn decode_responses_body(headers: &HeaderMap, body: Bytes) -> Result Err(GatewayError { - upstream_response: None, - status: StatusCode::UNSUPPORTED_MEDIA_TYPE, - code: "invalid_request_error", - message: "supported content encodings: identity, zstd".into(), - }), + .map_err(GatewayError::internal)??, + _ => { + return Err(GatewayError { + upstream_response: None, + status: StatusCode::UNSUPPORTED_MEDIA_TYPE, + code: "invalid_request_error", + message: "supported content encodings: identity, zstd".into(), + }); + } + }; + let raw: Box = + serde_json::from_slice(&body).map_err(|e| GatewayError::invalid(e.to_string()))?; + let body: Value = + serde_json::from_str(raw.get()).map_err(|e| GatewayError::invalid(e.to_string()))?; + if !body.is_object() || body["stream"] != true || !body["model"].is_string() { + return Err(GatewayError::invalid( + "native endpoint requires an object with model and stream: true", + )); } + let forwarded = forward_request_headers(&headers); + let usage = crate::telemetry::RequestTelemetry::new(gateway.telemetry.clone(), &headers).start( + body["model"].as_str().unwrap_or("unknown"), + "responses", + "http", + ); + respond(gateway, None, Some((raw, forwarded)), true, false, usage).await } pub(crate) fn forward_request_headers(headers: &HeaderMap) -> HeaderMap { @@ -549,11 +546,6 @@ async fn compact( let telemetry = crate::telemetry::RequestTelemetry::new(gateway.telemetry.clone(), &headers); let headers = crate::headers::request_headers(&headers); let Json(input) = body.map_err(|e| GatewayError::invalid(e.body_text()))?; - if input.client_metadata.is_some() { - return Err(GatewayError::invalid( - "client_metadata is not supported for compact", - )); - } let model = gateway .models .iter() diff --git a/tests/cases/responses_limits.rs b/tests/cases/responses_limits.rs deleted file mode 100644 index e9d5211..0000000 --- a/tests/cases/responses_limits.rs +++ /dev/null @@ -1,322 +0,0 @@ -use super::*; - -const HTTP_LIMIT: usize = 64 * 1024 * 1024; - -#[tokio::test] -async fn public_http_streaming_metadata_keeps_native_selection() { - let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(5)).await; - let response = h - .request(json!({ - "stream":true, - "input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"hello"}]}], - "client_metadata":{"trace":"keep"} - })) - .send() - .await - .unwrap(); - assert_eq!(response.status(), StatusCode::OK); - assert!( - response - .text() - .await - .unwrap() - .contains("response.completed") - ); - let received = h.fake.received.lock().unwrap(); - assert_eq!(received.len(), 1); - assert_eq!(received[0].1["client_metadata"]["trace"], "keep"); - assert_eq!(received[0].1["input"].as_array().unwrap().len(), 1); - assert_eq!(received[0].1["input"][0]["role"], "user"); -} - -#[tokio::test] -async fn native_http_preserves_done_items_when_terminal_output_is_empty() { - let mut fixture = events(); - *fixture.last_mut().unwrap() = completed(vec![]); - let h = Harness::new( - fixture.clone(), - StatusCode::OK, - false, - Duration::from_secs(5), - ) - .await; - let response = reqwest::Client::new() - .post(format!("{}/backend-api/codex/responses", h.url)) - .bearer_auth("client-key") - .json(&json!({"model":h.model,"stream":true,"input":[]})) - .send() - .await - .unwrap(); - assert_eq!(response.status(), StatusCode::OK); - assert_eq!(parse_sse(&response.text().await.unwrap()), fixture); -} - -#[tokio::test] -async fn responses_accept_large_images_over_http_with_identity_and_zstd() { - for path in [ - "/v1/responses", - "/codex/responses", - "/backend-api/codex/responses", - ] { - for compressed in [false, true] { - let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(30)).await; - let image = format!("data:image/png;base64,{}", "A".repeat(17 * 1024 * 1024)); - let body = json!({"model":h.model,"stream":true,"input":[{ - "type":"message","role":"user","content":[{"type":"input_image","image_url":image}] - }]}); - let wire = serde_json::to_vec(&body).unwrap(); - let mut request = reqwest::Client::new() - .post(format!("{}{path}", h.url)) - .bearer_auth("client-key") - .header("content-type", "application/json"); - let wire = if compressed { - request = request.header("content-encoding", "zstd"); - zstd::encode_all(wire.as_slice(), 1).unwrap() - } else { - wire - }; - let response = request.body(wire).send().await.unwrap(); - assert_eq!( - response.status(), - StatusCode::OK, - "{path}, compressed={compressed}" - ); - assert!( - response - .text() - .await - .unwrap() - .contains("response.completed") - ); - let received = h.fake.received.lock().unwrap(); - assert_eq!(received.len(), 1); - let input = received[0].1["input"].as_array().unwrap(); - let users: Vec<_> = input.iter().filter(|item| item["role"] == "user").collect(); - assert_eq!(users.len(), 1); - assert_eq!(users[0]["content"].as_array().unwrap().len(), 1); - assert!(users[0]["content"][0]["image_url"] == image); - if path == "/v1/responses" && !compressed { - assert_eq!(input[0]["role"], "developer"); - } else { - assert_eq!(input.len(), 1); - } - } - } -} - -#[tokio::test] -async fn responses_http_enforces_wire_and_decoded_boundaries() { - for path in [ - "/v1/responses", - "/codex/responses", - "/backend-api/codex/responses", - ] { - for compressed in [false, true] { - let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(30)).await; - for size in [HTTP_LIMIT, HTTP_LIMIT + 1] { - let mut wire = serde_json::to_vec(&json!({"model":h.model,"stream":true,"input":[{ - "type":"message","role":"user","content":[{"type":"input_text","text":"hello"}] - }]})) - .unwrap(); - wire.resize(size, b' '); - let mut request = reqwest::Client::new() - .post(format!("{}{path}", h.url)) - .bearer_auth("client-key") - .header("content-type", "application/json"); - let wire = if compressed { - request = request.header("content-encoding", "zstd"); - zstd::encode_all(wire.as_slice(), 1).unwrap() - } else { - wire - }; - let response = request.body(wire).send().await.unwrap(); - let expected = if size == HTTP_LIMIT { - StatusCode::OK - } else { - StatusCode::PAYLOAD_TOO_LARGE - }; - assert_eq!( - response.status(), - expected, - "{path}, compressed={compressed}, size={size}" - ); - let text = response.text().await.unwrap(); - if size == HTTP_LIMIT { - assert!(text.contains("response.completed")); - } - } - assert_eq!(h.fake.received.lock().unwrap().len(), 1); - } - } -} - -#[tokio::test] -async fn responses_preserves_public_metadata_without_accepting_continuation_or_unknown_fields() { - let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(5)).await; - let metadata = json!({"trace":"keep","ws_request_header_x_codex_turn_metadata":"turn"}); - let response = h - .request(json!({"input":"hello","client_metadata":metadata})) - .send() - .await - .unwrap(); - assert_eq!(response.status(), StatusCode::OK); - assert_eq!( - response.json::().await.unwrap()["status"], - "completed" - ); - assert_eq!( - h.fake.received.lock().unwrap()[0].1["client_metadata"], - metadata - ); - for field in [ - json!({"client_metadata":{"trace":1}}), - json!({"client_metadata":null}), - json!({"previous_response_id":"resp_1"}), - json!({"stream_id":"lane"}), - json!({"unknown_field":true}), - ] { - let mut body = field; - body["input"] = json!("hello"); - let response = h.request(body).send().await.unwrap(); - assert!(response.status().is_client_error()); - } - assert_eq!(h.fake.received.lock().unwrap().len(), 1); -} - -#[tokio::test] -async fn responses_compression_validates_encoding_and_public_content_type() { - let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(5)).await; - for path in [ - "/v1/responses", - "/codex/responses", - "/backend-api/codex/responses", - ] { - for (encoding, status) in [ - ("gzip", StatusCode::UNSUPPORTED_MEDIA_TYPE), - ("zstd", StatusCode::BAD_REQUEST), - ] { - let response = reqwest::Client::new() - .post(format!("{}{path}", h.url)) - .bearer_auth("client-key") - .header("content-type", "application/json") - .header("content-encoding", encoding) - .body("invalid compressed body") - .send() - .await - .unwrap(); - assert_eq!(response.status(), status); - } - } - let response = reqwest::Client::new() - .post(format!("{}/v1/responses", h.url)) - .bearer_auth("client-key") - .body(json!({"model":h.model,"input":"hello"}).to_string()) - .send() - .await - .unwrap(); - assert_eq!(response.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE); - assert!(h.fake.received.lock().unwrap().is_empty()); -} - -#[tokio::test] -async fn responses_rejects_truncated_zstd_and_oversized_compressed_wire() { - use std::io::Write; - - let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(5)).await; - let body = serde_json::to_vec(&json!({"model":h.model,"stream":true,"input":"hello"})).unwrap(); - let mut encoder = zstd::Encoder::new(Vec::new(), 1).unwrap(); - encoder.include_checksum(true).unwrap(); - encoder.write_all(&body).unwrap(); - let wire = encoder.finish().unwrap(); - let mut truncated = wire.clone(); - truncated.pop(); - // Skippable padding grows only wire bytes, isolating the compressed ingress cap. - let mut oversized = wire; - oversized.extend_from_slice(&0x184D_2A50u32.to_le_bytes()); - let padding_len = HTTP_LIMIT + 1 - oversized.len() - 4; - oversized.extend_from_slice(&(padding_len as u32).to_le_bytes()); - oversized.resize(HTTP_LIMIT + 1, 0); - assert_eq!(zstd::decode_all(oversized.as_slice()).unwrap(), body); - for path in [ - "/v1/responses", - "/codex/responses", - "/backend-api/codex/responses", - ] { - for (wire, status) in [ - (&truncated, StatusCode::BAD_REQUEST), - (&oversized, StatusCode::PAYLOAD_TOO_LARGE), - ] { - let response = reqwest::Client::new() - .post(format!("{}{path}", h.url)) - .bearer_auth("client-key") - .header("content-type", "application/json") - .header("content-encoding", "zstd") - .body(wire.clone()) - .send() - .await - .unwrap(); - assert_eq!(response.status(), status, "{path}"); - } - } - assert!(h.fake.received.lock().unwrap().is_empty()); -} - -#[tokio::test] -async fn compact_rejects_response_metadata_including_null() { - let h = Harness::new( - vec![completed(vec![ - json!({"type":"compaction","id":"cmp_1","encrypted_content":"fixture"}), - ])], - StatusCode::OK, - false, - Duration::from_secs(5), - ) - .await; - for path in ["/v1/responses/compact", "/codex/responses/compact"] { - for metadata in [json!({"trace":"keep"}), json!({}), Value::Null] { - let response = reqwest::Client::new() - .post(format!("{}{path}", h.url)) - .bearer_auth("client-key") - .json(&json!({"model":h.model,"input":"hello","client_metadata":metadata})) - .send() - .await - .unwrap(); - assert_eq!(response.status(), StatusCode::BAD_REQUEST, "{path}"); - } - let duplicated = format!( - r#"{{"model":"{}","input":"hello","input":"duplicate"}}"#, - h.model - ); - let response = reqwest::Client::new() - .post(format!("{}{path}", h.url)) - .bearer_auth("client-key") - .header("content-type", "application/json") - .body(duplicated) - .send() - .await - .unwrap(); - assert_eq!(response.status(), StatusCode::BAD_REQUEST); - } - assert!(h.fake.received.lock().unwrap().is_empty()); -} - -#[tokio::test] -async fn responses_larger_budget_does_not_expand_unrelated_endpoints() { - let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(5)).await; - for (path, status) in [ - ("/v1/responses/compact", StatusCode::BAD_REQUEST), - ("/v1/images/generations", StatusCode::PAYLOAD_TOO_LARGE), - ] { - let response = reqwest::Client::new() - .post(format!("{}{path}", h.url)) - .bearer_auth("client-key") - .header("content-type", "application/json") - .body(vec![b' '; 16 * 1024 * 1024 + 1]) - .send() - .await - .unwrap(); - assert_eq!(response.status(), status, "{path}"); - assert!(response.text().await.unwrap().contains("length limit")); - } - assert!(h.fake.received.lock().unwrap().is_empty()); -} diff --git a/tests/gateway.rs b/tests/gateway.rs index a914a1d..1fe0bd3 100644 --- a/tests/gateway.rs +++ b/tests/gateway.rs @@ -2,8 +2,6 @@ mod file_cases; #[path = "cases/native.rs"] mod native_cases; -#[path = "cases/responses_limits.rs"] -mod responses_limit_cases; mod support; use axum::{ Router, @@ -225,8 +223,8 @@ impl Harness { .route( "/responses", post(upstream) - .get(upstream_ws) - .layer(axum::extract::DefaultBodyLimit::max(65 * 1024 * 1024)), + .layer(axum::extract::DefaultBodyLimit::max(65 * 1024 * 1024)) + .get(upstream_ws), ) .route("/realtime/calls", post(realtime_cases::upstream_call)) .route("/realtime", axum::routing::get(upstream_ws)) @@ -906,6 +904,14 @@ async fn native_rejects_unauthorized_and_nonstream_and_decompression_bombs() { .status(), StatusCode::PAYLOAD_TOO_LARGE ); + let oversized = client + .post(&url) + .bearer_auth("client-key") + .body(vec![b' '; 64 * 1024 * 1024 + 1]) + .send() + .await + .unwrap(); + assert_eq!(oversized.status(), StatusCode::PAYLOAD_TOO_LARGE); assert!(h.fake.received.lock().unwrap().is_empty()); } @@ -1590,14 +1596,17 @@ mod response_type_cases; #[tokio::test] async fn public_http_preserves_prepared_codex_requests() { - let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(5)).await; + let h = Harness::new(events(), StatusCode::OK, false, Duration::from_secs(30)).await; + let image = format!("data:image/png;base64,{}", "A".repeat(17 * 1024 * 1024)); let body = json!({"model":"future-model", "stream":true, "store":false, - "instructions":"unchanged", "input":[], "tools":null, + "instructions":"unchanged", "tools":null, + "input":[{"role":"user","content":[{"type":"input_image","image_url":image}]}], "client_metadata":{"trace":"keep"}, "future_field":{"keep":true}}); - let raw = serde_json::to_vec(&body).unwrap(); - for compressed in [false, true] { + let mut raw = serde_json::to_vec(&body).unwrap(); + raw.resize(64 * 1024 * 1024, b' '); + for (path, compressed) in [("v1", false), ("codex", true), ("backend-api/codex", false)] { let response = reqwest::Client::new() - .post(format!("{}/v1/responses", h.url)) + .post(format!("{}/{path}/responses", h.url)) .bearer_auth("client-key") .header("content-type", "application/json") .header( @@ -1622,9 +1631,9 @@ async fn public_http_preserves_prepared_codex_requests() { ); } let received = h.fake.received.lock().unwrap(); - assert_eq!(received.len(), 2); + assert_eq!(received.len(), 3); for (_, forwarded) in received.iter() { - assert_eq!(forwarded, &body); + assert!(forwarded == &body); } }