diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 60bd57a94..13b977ea2 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -825,6 +825,7 @@ dependencies = [ "windows-sys 0.61.2", "winreg 0.52.0", "zip 2.4.2", + "zstd", ] [[package]] diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 9f7841511..9fc61c123 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -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" diff --git a/src-tauri/src/proxy/content_encoding.rs b/src-tauri/src/proxy/content_encoding.rs new file mode 100644 index 000000000..970ffd289 --- /dev/null +++ b/src-tauri/src/proxy/content_encoding.rs @@ -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>, 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 登录态对请求体启用 zstd(Compression::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.4:codings 按**应用顺序**列出,故解压须**反向**(最后应用的先解)。 +/// 返回 `Ok(None)` 表示存在不受支持的编码、原样透传——此时调用方必须保留 +/// content-encoding 头,否则下游(诊断 / 客户端)会把压缩字节误当明文。 +pub(crate) fn decompress_body( + content_encoding: &str, + body: &[u8], +) -> Result>, 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> = 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 { + let combined = headers + .get_all("content-encoding") + .iter() + .filter_map(|v| v.to_str().ok()) + .map(str::trim) + .filter(|s| !s.is_empty()) + .collect::>() + .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); + } +} diff --git a/src-tauri/src/proxy/forwarder.rs b/src-tauri/src/proxy/forwarder.rs index 0c840c399..d11e52901 100644 --- a/src-tauri/src/proxy/forwarder.rs +++ b/src-tauri/src/proxy/forwarder.rs @@ -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, diff --git a/src-tauri/src/proxy/handlers.rs b/src-tauri/src/proxy/handlers.rs index c399f1ba0..35c3062ee 100644 --- a/src-tauri/src/proxy/handlers.rs +++ b/src-tauri/src/proxy/handlers.rs @@ -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 { + 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}")))?; diff --git a/src-tauri/src/proxy/http_client.rs b/src-tauri/src/proxy/http_client.rs index 1d3ddb2c9..97e927e46 100644 --- a/src-tauri/src/proxy/http_client.rs +++ b/src-tauri/src/proxy/http_client.rs @@ -223,7 +223,8 @@ fn build_client(proxy_url: Option<&str>) -> Result { // 响应解压由 response_processor 根据 content-encoding 手动处理。 .no_gzip() .no_brotli() - .no_deflate(); + .no_deflate() + .no_zstd(); // 有代理地址则使用代理,否则跟随系统代理 if let Some(url) = proxy_url { diff --git a/src-tauri/src/proxy/mod.rs b/src-tauri/src/proxy/mod.rs index 585703c63..cbc24404c 100644 --- a/src-tauri/src/proxy/mod.rs +++ b/src-tauri/src/proxy/mod.rs @@ -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; diff --git a/src-tauri/src/proxy/response_processor.rs b/src-tauri/src/proxy/response_processor.rs index 59d525b1b..238e14420 100644 --- a/src-tauri/src/proxy/response_processor.rs +++ b/src-tauri/src/proxy/response_processor.rs @@ -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>, 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 { - 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!(