fix: decompress body before forward and support zstd (#3817)

* fix(proxy): decompress Codex request body before forward, support zstd

Codex Desktop sends zstd-compressed request bodies when authenticated
against the Codex backend, which broke local proxy routing because the
handlers parsed the raw bytes with serde_json directly.

Reworked on top of current main so it preserves the response_processor
behavior that landed after this PR was first opened:

- Extract content-encoding helpers into a shared proxy::content_encoding
  module. decompress_body keeps returning Option<Vec<u8>> so unknown
  encodings stay pass-through with their content-encoding header intact,
  and keeps the deflate zlib-then-raw fallback (RFC 9110).
- Add zstd/zst support (zstd 0.13) and disable reqwest's auto zstd
  decompression via .no_zstd() for parity with gzip/br/deflate.
- Decompress the request body before JSON parsing in the three Codex
  handlers (chat_completions / responses / responses_compact) and strip
  the stale content-encoding / content-length / transfer-encoding headers
  so the forwarder regenerates them.
- Support stacked codings (e.g. "gzip, zstd") by decoding in reverse
  order and merge repeated Content-Encoding headers via get_all.

Fixes #3764
Fixes #3696

Co-authored-by: chenx-dust <16610294+chenx-dust@users.noreply.github.com>

* fix(proxy): decompress upstream error bodies before reading them

The forwarder error branch consumes non-2xx responses via String::from_utf8
directly, bypassing read_decoded_body. reqwest has no auto-decompression
feature enabled, so a compressed error body (gzip/br/deflate/zstd) arrives
as raw bytes, fails from_utf8, and gets dropped, hiding upstream rate-limit
and auth details from the client.

Decode the error body with the shared proxy::content_encoding helper,
mirroring the success path. Falls back to the raw bytes when the encoding is
unsupported or decoding fails.

Co-authored-by: chenx-dust <16610294+chenx-dust@users.noreply.github.com>

---------

Co-authored-by: Jason <farion1231@gmail.com>
Co-authored-by: chenx-dust <16610294+chenx-dust@users.noreply.github.com>
This commit is contained in:
Chenx Dust
2026-06-26 08:50:59 +08:00
committed by GitHub
parent 524b9d9825
commit 1a0e8c7a44
8 changed files with 306 additions and 92 deletions
+1
View File
@@ -825,6 +825,7 @@ dependencies = [
"windows-sys 0.61.2",
"winreg 0.52.0",
"zip 2.4.2",
"zstd",
]
[[package]]
+1
View File
@@ -43,6 +43,7 @@ reqwest = { version = "0.12", features = ["rustls-tls", "json", "stream", "socks
arboard = "3.6"
flate2 = "1"
brotli = "7"
zstd = "0.13"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "sync"] }
futures = "0.3"
async-stream = "0.3"
+234
View File
@@ -0,0 +1,234 @@
//! HTTP content-encoding 工具。
//!
//! reqwest 的自动解压已禁用(为了透传 accept-encoding),需要手动解压。
//! 请求侧(如 Codex Desktop 在登录态发压缩请求体)与响应侧(上游压缩响应体)
//! 共用同一套解压逻辑。
use axum::http::header::HeaderMap;
use std::io::Read;
/// 把 content-encoding 值拆成有序 coding 列表(去掉 identity 与空值)。
///
/// HTTP 允许堆叠编码(如 `gzip, zstd`),各 coding 以逗号分隔;亦允许重复
/// content-encoding 头,语义等同逗号拼接(见 [`get_content_encoding`])。
fn split_codings(content_encoding: &str) -> Vec<&str> {
content_encoding
.split(',')
.map(str::trim)
.filter(|c| !c.is_empty() && *c != "identity")
.collect()
}
/// 单个 coding 是否可被解压。
fn is_single_supported(coding: &str) -> bool {
matches!(
coding,
"gzip" | "x-gzip" | "deflate" | "br" | "zstd" | "zst"
)
}
/// 解压单个 content-coding。未知编码返回 `Ok(None)`。
fn decompress_single(coding: &str, body: &[u8]) -> Result<Option<Vec<u8>>, std::io::Error> {
match coding {
"gzip" | "x-gzip" => {
let mut decoder = flate2::read::GzDecoder::new(body);
let mut decompressed = Vec::new();
decoder.read_to_end(&mut decompressed)?;
Ok(Some(decompressed))
}
"deflate" => {
// RFC 9110: deflate 指 zlib 包裹格式;但部分上游 / 客户端发 raw deflate 流。
// 先按规范尝试 zlib,失败再回退 raw —— 否则合规来源必然解压失败,
// 原始压缩字节会被 fail-open 透传给 JSON 解析(#2234 形态 C 之一)。
let mut decompressed = Vec::new();
let mut zlib = flate2::read::ZlibDecoder::new(body);
match zlib.read_to_end(&mut decompressed) {
Ok(_) => Ok(Some(decompressed)),
Err(zlib_err) => {
log::debug!("deflate 按 zlib 解压失败({zlib_err}),回退 raw deflate");
let mut decompressed = Vec::new();
let mut raw = flate2::read::DeflateDecoder::new(body);
raw.read_to_end(&mut decompressed)?;
Ok(Some(decompressed))
}
}
}
"br" => {
let mut decompressed = Vec::new();
brotli::BrotliDecompress(&mut std::io::Cursor::new(body), &mut decompressed)?;
Ok(Some(decompressed))
}
"zstd" | "zst" => {
// Codex 登录态对请求体启用 zstdCompression::Zstd);上游也可能 zstd 压缩响应。
let decompressed = zstd::stream::decode_all(std::io::Cursor::new(body))?;
Ok(Some(decompressed))
}
_ => Ok(None),
}
}
/// 根据 content-encoding 解压 body 字节,支持堆叠编码(如 `gzip, zstd`)。
///
/// RFC 9110 §8.4codings 按**应用顺序**列出,故解压须**反向**(最后应用的先解)。
/// 返回 `Ok(None)` 表示存在不受支持的编码、原样透传——此时调用方必须保留
/// content-encoding 头,否则下游(诊断 / 客户端)会把压缩字节误当明文。
pub(crate) fn decompress_body(
content_encoding: &str,
body: &[u8],
) -> Result<Option<Vec<u8>>, std::io::Error> {
let codings = split_codings(content_encoding);
if codings.is_empty() {
return Ok(None);
}
// 任一 coding 不支持就整体放弃解压、保头透传,避免半解码的脏数据。
if !codings.iter().all(|c| is_single_supported(c)) {
log::warn!("不支持的 content-encoding: {content_encoding},跳过解压");
return Ok(None);
}
// 反向解码:列表末尾是最后应用的编码,须最先解。
let mut data: Option<Vec<u8>> = None;
for coding in codings.iter().rev() {
let input = data.as_deref().unwrap_or(body);
match decompress_single(coding, input)? {
Some(decompressed) => data = Some(decompressed),
// 上面 is_single_supported 已校验,理论不会发生;防御性兜底。
None => return Ok(None),
}
}
Ok(data)
}
/// 该 content-encoding(含堆叠,如 `gzip, zstd`)是否全部可被解压。
///
/// 请求侧用它做闸门:无法解压的压缩体不能透传给 JSON 解析,需直接拒绝。
pub(crate) fn is_supported_content_encoding(content_encoding: &str) -> bool {
let codings = split_codings(content_encoding);
!codings.is_empty() && codings.iter().all(|c| is_single_supported(c))
}
/// 从 header 提取 content-encoding(合并重复头,忽略 identity 与空值)。
///
/// HTTP 允许重复 content-encoding 头,语义等同逗号拼接,故用 `get_all` 合并;
/// 返回值可能含多个逗号分隔的 coding,交由 [`decompress_body`] 反向解码。
pub(crate) fn get_content_encoding(headers: &HeaderMap) -> Option<String> {
let combined = headers
.get_all("content-encoding")
.iter()
.filter_map(|v| v.to_str().ok())
.map(str::trim)
.filter(|s| !s.is_empty())
.collect::<Vec<_>>()
.join(", ")
.to_lowercase();
if split_codings(&combined).is_empty() {
return None;
}
Some(combined)
}
#[cfg(test)]
mod tests {
use super::*;
use axum::http::HeaderValue;
#[test]
fn decompress_body_deflate_handles_zlib_wrapped_per_rfc9110() {
// RFC 9110 规范的 deflate = zlib 包裹格式(合规来源发的就是这个)
let payload = br#"{"ok":true}"#;
let mut encoder =
flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
std::io::Write::write_all(&mut encoder, payload).unwrap();
let compressed = encoder.finish().unwrap();
let decompressed = decompress_body("deflate", &compressed).unwrap().unwrap();
assert_eq!(decompressed, payload);
}
#[test]
fn decompress_body_deflate_falls_back_to_raw_stream() {
// 部分来源违规发 raw deflate 流,保持兼容
let payload = br#"{"ok":true}"#;
let mut encoder =
flate2::write::DeflateEncoder::new(Vec::new(), flate2::Compression::default());
std::io::Write::write_all(&mut encoder, payload).unwrap();
let compressed = encoder.finish().unwrap();
let decompressed = decompress_body("deflate", &compressed).unwrap().unwrap();
assert_eq!(decompressed, payload);
}
#[test]
fn decompress_body_zstd_roundtrip() {
// Codex 登录态发的就是 zstd 压缩请求体
let payload = br#"{"hello":"world","n":42}"#;
let compressed = zstd::stream::encode_all(std::io::Cursor::new(&payload[..]), 0).unwrap();
let decompressed = decompress_body("zstd", &compressed).unwrap().unwrap();
assert_eq!(decompressed, payload);
}
#[test]
fn decompress_body_stacked_gzip_then_zstd_decodes_in_reverse() {
// Content-Encoding: gzip, zstd 表示先 gzip 后 zstd,解压须反向(先 zstd 后 gzip)
let payload = br#"{"stacked":true}"#;
let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
std::io::Write::write_all(&mut gz, payload).unwrap();
let gzipped = gz.finish().unwrap();
let stacked = zstd::stream::encode_all(std::io::Cursor::new(&gzipped[..]), 0).unwrap();
let decompressed = decompress_body("gzip, zstd", &stacked).unwrap().unwrap();
assert_eq!(decompressed, payload);
}
#[test]
fn decompress_body_stacked_with_unsupported_returns_none() {
// 堆叠里只要有一个不支持,就整体保头透传
let result = decompress_body("snappy, zstd", b"\x00\x01\x02\x03").unwrap();
assert!(result.is_none());
}
#[test]
fn decompress_body_unknown_encoding_returns_none_to_keep_headers() {
// 未知编码必须返回 None(而非伪装成"已解码"),否则 content-encoding
// 头被剥掉,下游诊断会把压缩字节误报成明文
let result = decompress_body("snappy", b"\x00\x01\x02\x03").unwrap();
assert!(result.is_none());
}
#[test]
fn is_supported_content_encoding_matches_decompressable() {
for enc in [
"gzip",
"x-gzip",
"deflate",
"br",
"zstd",
"zst",
"gzip, zstd",
] {
assert!(is_supported_content_encoding(enc), "{enc} 应受支持");
}
for enc in ["identity", "snappy", "compress", "", "gzip, snappy"] {
assert!(!is_supported_content_encoding(enc), "{enc} 不应受支持");
}
}
#[test]
fn get_content_encoding_combines_repeated_headers() {
// 重复的 content-encoding 头等同逗号拼接,须用 get_all 合并
let mut headers = HeaderMap::new();
headers.append("content-encoding", HeaderValue::from_static("gzip"));
headers.append("content-encoding", HeaderValue::from_static("zstd"));
assert_eq!(
get_content_encoding(&headers).as_deref(),
Some("gzip, zstd")
);
}
#[test]
fn get_content_encoding_ignores_identity_only() {
let mut headers = HeaderMap::new();
headers.append("content-encoding", HeaderValue::from_static("identity"));
assert_eq!(get_content_encoding(&headers), None);
}
}
+15 -1
View File
@@ -5,6 +5,7 @@
use super::hyper_client::ProxyResponse;
use super::{
body_filter::filter_private_params_with_whitelist,
content_encoding::{decompress_body, get_content_encoding},
error::*,
failover_switch::FailoverSwitchManager,
json_canonical::{canonicalize_value, short_value_hash},
@@ -1965,7 +1966,20 @@ impl RequestForwarder {
Ok((response, resolved_claude_api_format, outbound_model))
} else {
let status_code = status.as_u16();
let body_text = String::from_utf8(response.bytes().await?.to_vec()).ok();
// 错误响应同样可能被上游压缩(content-encoding)。reqwest 未启用任何
// 自动解压 feature,这里拿到的是原始字节;不解压的话,压缩过的错误体会
// 在 from_utf8 处变成非 UTF-8 而被丢弃,隐藏掉上游的限流/鉴权等详情。
let encoding = get_content_encoding(response.headers());
let raw = response.bytes().await?;
let decoded = match encoding {
Some(encoding) => match decompress_body(&encoding, &raw) {
Ok(Some(decompressed)) => decompressed,
// 不支持的编码 / 解压失败:退回原始字节,尽量保留可读信息
_ => raw.to_vec(),
},
None => raw.to_vec(),
};
let body_text = String::from_utf8(decoded).ok();
Err(ProxyError::UpstreamError {
status: status_code,
+50 -3
View File
@@ -8,6 +8,7 @@
//! - Claude 的格式转换逻辑保留在此文件(用于 OpenRouter 旧接口回退)
use super::{
content_encoding::{decompress_body, get_content_encoding, is_supported_content_encoding},
error_mapper::{get_error_message, map_proxy_error_to_status},
forwarder::ActiveConnectionGuard,
handler_config::{
@@ -568,6 +569,49 @@ fn endpoint_with_query(uri: &axum::http::Uri, endpoint: &str) -> String {
}
}
/// Codex 客户端(尤其 Desktop 登录态)可能对请求体启用 zstd 压缩,使得后续
/// `serde_json::from_slice` 直接解析失败。这里在解析前解压,并剥掉已失真的实体头
/// content-encoding / content-length / transfer-encoding)——转发层会基于解压后的
/// 明文 JSON 重新生成正确的头。
fn decode_codex_request_body(
headers: &mut axum::http::HeaderMap,
body_bytes: Bytes,
) -> Result<Bytes, ProxyError> {
let Some(encoding) = get_content_encoding(headers) else {
return Ok(body_bytes);
};
if !is_supported_content_encoding(&encoding) {
return Err(ProxyError::InvalidRequest(format!(
"Unsupported request content-encoding: {encoding}"
)));
}
log::debug!("[Codex] 解压请求体: content-encoding={encoding}");
let decompressed = match decompress_body(&encoding, &body_bytes) {
Ok(Some(decompressed)) => decompressed,
// is_supported_content_encoding 已确保编码受支持,正常不会返回 None;
// 防御性兜底:宁可报错,也不能把压缩字节当 JSON 透传下去。
Ok(None) => {
return Err(ProxyError::InvalidRequest(format!(
"Unsupported request content-encoding: {encoding}"
)));
}
Err(e) => {
log::warn!("[Codex] 请求体解压失败 ({encoding}): {e}");
return Err(ProxyError::InvalidRequest(format!(
"Failed to decompress request body ({encoding}): {e}"
)));
}
};
headers.remove(axum::http::header::CONTENT_ENCODING);
headers.remove(axum::http::header::CONTENT_LENGTH);
headers.remove(axum::http::header::TRANSFER_ENCODING);
Ok(Bytes::from(decompressed))
}
// ============================================================================
// Codex API 处理器
// ============================================================================
@@ -580,13 +624,14 @@ pub async fn handle_chat_completions(
let (parts, req_body) = request.into_parts();
let method = parts.method.clone();
let uri = parts.uri;
let headers = parts.headers;
let mut headers = parts.headers;
let extensions = parts.extensions;
let body_bytes = req_body
.collect()
.await
.map_err(|e| ProxyError::Internal(format!("Failed to read request body: {e}")))?
.to_bytes();
let body_bytes = decode_codex_request_body(&mut headers, body_bytes)?;
let body: Value = serde_json::from_slice(&body_bytes)
.map_err(|e| ProxyError::Internal(format!("Failed to parse request body: {e}")))?;
@@ -645,13 +690,14 @@ pub async fn handle_responses(
let (parts, req_body) = request.into_parts();
let method = parts.method.clone();
let uri = parts.uri;
let headers = parts.headers;
let mut headers = parts.headers;
let extensions = parts.extensions;
let body_bytes = req_body
.collect()
.await
.map_err(|e| ProxyError::Internal(format!("Failed to read request body: {e}")))?
.to_bytes();
let body_bytes = decode_codex_request_body(&mut headers, body_bytes)?;
let body: Value = serde_json::from_slice(&body_bytes)
.map_err(|e| ProxyError::Internal(format!("Failed to parse request body: {e}")))?;
@@ -723,13 +769,14 @@ pub async fn handle_responses_compact(
let (parts, req_body) = request.into_parts();
let method = parts.method.clone();
let uri = parts.uri;
let headers = parts.headers;
let mut headers = parts.headers;
let extensions = parts.extensions;
let body_bytes = req_body
.collect()
.await
.map_err(|e| ProxyError::Internal(format!("Failed to read request body: {e}")))?
.to_bytes();
let body_bytes = decode_codex_request_body(&mut headers, body_bytes)?;
let body: Value = serde_json::from_slice(&body_bytes)
.map_err(|e| ProxyError::Internal(format!("Failed to parse request body: {e}")))?;
+2 -1
View File
@@ -223,7 +223,8 @@ fn build_client(proxy_url: Option<&str>) -> Result<Client, String> {
// 响应解压由 response_processor 根据 content-encoding 手动处理。
.no_gzip()
.no_brotli()
.no_deflate();
.no_deflate()
.no_zstd();
// 有代理地址则使用代理,否则跟随系统代理
if let Some(url) = proxy_url {
+1
View File
@@ -5,6 +5,7 @@
pub mod body_filter;
pub mod cache_injector;
pub mod circuit_breaker;
pub(crate) mod content_encoding;
pub mod copilot_optimizer;
pub mod error;
pub mod error_mapper;
+2 -87
View File
@@ -3,6 +3,7 @@
//! 统一处理流式和非流式 API 响应
use super::{
content_encoding::{decompress_body, get_content_encoding},
forwarder::ActiveConnectionGuard,
handler_config::{StreamUsageEventFilter, UsageParserConfig},
handler_context::{RequestContext, StreamingTimeoutConfig},
@@ -19,7 +20,6 @@ use bytes::Bytes;
use futures::stream::{Stream, StreamExt};
use serde_json::Value;
use std::{
io::Read,
sync::{
atomic::{AtomicBool, Ordering},
Arc,
@@ -29,60 +29,9 @@ use std::{
use tokio::sync::Mutex;
// ============================================================================
// 响应解压
// 响应头处理
// ============================================================================
/// 根据 content-encoding 解压响应体字节
///
/// reqwest 自动解压已禁用(为了透传 accept-encoding),需要手动解压。
/// 返回 `Ok(None)` 表示编码不受支持、原样透传——此时调用方必须保留
/// content-encoding 头,否则下游(诊断/客户端)会把压缩字节误当明文。
fn decompress_body(content_encoding: &str, body: &[u8]) -> Result<Option<Vec<u8>>, std::io::Error> {
match content_encoding {
"gzip" | "x-gzip" => {
let mut decoder = flate2::read::GzDecoder::new(body);
let mut decompressed = Vec::new();
decoder.read_to_end(&mut decompressed)?;
Ok(Some(decompressed))
}
"deflate" => {
// RFC 9110: deflate 指 zlib 包裹格式;但部分上游发 raw deflate 流。
// 先按规范尝试 zlib,失败再回退 raw —— 否则合规上游必然解压失败,
// 原始压缩字节会被 fail-open 透传给 JSON 解析(#2234 形态 C 之一)。
let mut decompressed = Vec::new();
let mut zlib = flate2::read::ZlibDecoder::new(body);
match zlib.read_to_end(&mut decompressed) {
Ok(_) => Ok(Some(decompressed)),
Err(zlib_err) => {
log::debug!("deflate 按 zlib 解压失败({zlib_err}),回退 raw deflate");
let mut decompressed = Vec::new();
let mut raw = flate2::read::DeflateDecoder::new(body);
raw.read_to_end(&mut decompressed)?;
Ok(Some(decompressed))
}
}
}
"br" => {
let mut decompressed = Vec::new();
brotli::BrotliDecompress(&mut std::io::Cursor::new(body), &mut decompressed)?;
Ok(Some(decompressed))
}
_ => {
log::warn!("未知的 content-encoding: {content_encoding},跳过解压");
Ok(None)
}
}
}
/// 从响应头提取 content-encoding(忽略 identity 和 chunked
fn get_content_encoding(headers: &HeaderMap) -> Option<String> {
headers
.get("content-encoding")
.and_then(|v| v.to_str().ok())
.map(|s| s.trim().to_lowercase())
.filter(|s| !s.is_empty() && s != "identity")
}
/// RFC 2616 / RFC 7230 中定义的不应被代理继续转发的响应头。
const HOP_BY_HOP_RESPONSE_HEADERS: &[&str] = &[
"connection",
@@ -878,40 +827,6 @@ mod tests {
use std::sync::Arc;
use tokio::sync::RwLock;
#[test]
fn decompress_body_deflate_handles_zlib_wrapped_per_rfc9110() {
// RFC 9110 规范的 deflate = zlib 包裹格式(合规上游发的就是这个)
let payload = br#"{"ok":true}"#;
let mut encoder =
flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
std::io::Write::write_all(&mut encoder, payload).unwrap();
let compressed = encoder.finish().unwrap();
let decompressed = decompress_body("deflate", &compressed).unwrap().unwrap();
assert_eq!(decompressed, payload);
}
#[test]
fn decompress_body_deflate_falls_back_to_raw_stream() {
// 部分上游违规发 raw deflate 流,保持兼容
let payload = br#"{"ok":true}"#;
let mut encoder =
flate2::write::DeflateEncoder::new(Vec::new(), flate2::Compression::default());
std::io::Write::write_all(&mut encoder, payload).unwrap();
let compressed = encoder.finish().unwrap();
let decompressed = decompress_body("deflate", &compressed).unwrap().unwrap();
assert_eq!(decompressed, payload);
}
#[test]
fn decompress_body_unknown_encoding_returns_none_to_keep_headers() {
// 未知编码必须返回 None(而非伪装成"已解码"),否则 content-encoding
// 头被剥掉,下游诊断会把压缩字节误报成明文
let result = decompress_body("zstd", b"\x28\xb5\x2f\xfd").unwrap();
assert!(result.is_none());
}
#[test]
fn test_strip_sse_field_accepts_optional_space() {
assert_eq!(