diff --git a/Cargo.lock b/Cargo.lock index 1d096996..4827b683 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3143,7 +3143,7 @@ dependencies = [ [[package]] name = "tw-api" -version = "0.64.0" +version = "0.64.1" dependencies = [ "serde", "serde_json", @@ -3154,7 +3154,7 @@ dependencies = [ [[package]] name = "tw-bedrock" -version = "0.64.0" +version = "0.64.1" dependencies = [ "aws-credential-types", "aws-sigv4", @@ -3175,7 +3175,7 @@ dependencies = [ [[package]] name = "tw-breaker" -version = "0.64.0" +version = "0.64.1" dependencies = [ "serde", "serde_json", @@ -3183,7 +3183,7 @@ dependencies = [ [[package]] name = "tw-config" -version = "0.64.0" +version = "0.64.1" dependencies = [ "blake3", "libc", @@ -3208,7 +3208,7 @@ dependencies = [ [[package]] name = "tw-control" -version = "0.64.0" +version = "0.64.1" dependencies = [ "axum", "base64", @@ -3250,7 +3250,7 @@ dependencies = [ [[package]] name = "tw-dialect" -version = "0.64.0" +version = "0.64.1" dependencies = [ "serde", "serde_json", @@ -3258,7 +3258,7 @@ dependencies = [ [[package]] name = "tw-engine" -version = "0.64.0" +version = "0.64.1" dependencies = [ "serde", "serde_json", @@ -3271,7 +3271,7 @@ dependencies = [ [[package]] name = "tw-gateway" -version = "0.64.0" +version = "0.64.1" dependencies = [ "arc-swap", "async-stream", @@ -3320,7 +3320,7 @@ dependencies = [ [[package]] name = "tw-guard" -version = "0.64.0" +version = "0.64.1" dependencies = [ "base64", "bytes", @@ -3335,7 +3335,7 @@ dependencies = [ [[package]] name = "tw-link" -version = "0.64.0" +version = "0.64.1" dependencies = [ "serde", "serde_json", @@ -3349,7 +3349,7 @@ dependencies = [ [[package]] name = "tw-observe" -version = "0.64.0" +version = "0.64.1" dependencies = [ "tokio", "tracing", @@ -3358,7 +3358,7 @@ dependencies = [ [[package]] name = "tw-plugin" -version = "0.64.0" +version = "0.64.1" dependencies = [ "libc", "rand 0.10.2", @@ -3373,7 +3373,7 @@ dependencies = [ [[package]] name = "tw-pricing" -version = "0.64.0" +version = "0.64.1" dependencies = [ "arc-swap", "flate2", @@ -3386,14 +3386,14 @@ dependencies = [ [[package]] name = "tw-secret" -version = "0.64.0" +version = "0.64.1" dependencies = [ "thiserror", ] [[package]] name = "tw-store" -version = "0.64.0" +version = "0.64.1" dependencies = [ "blake3", "bytes", @@ -3413,7 +3413,7 @@ dependencies = [ [[package]] name = "tw-types" -version = "0.64.0" +version = "0.64.1" dependencies = [ "serde", "serde_json", @@ -3422,7 +3422,7 @@ dependencies = [ [[package]] name = "tw-watch" -version = "0.64.0" +version = "0.64.1" dependencies = [ "notify", "tempfile", @@ -3432,7 +3432,7 @@ dependencies = [ [[package]] name = "tw-yaml" -version = "0.64.0" +version = "0.64.1" dependencies = [ "saphyr-parser", "serde_yaml_ng", @@ -3442,7 +3442,7 @@ dependencies = [ [[package]] name = "twcore" -version = "0.64.0" +version = "0.64.1" dependencies = [ "anyhow", "axum", diff --git a/Cargo.toml b/Cargo.toml index f91dda61..8391e8b6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -54,7 +54,7 @@ categories = ["network-programming", "web-programming::http-server"] # 二进制走 CalVer,crate 走 SemVer —— 两者是两回事,卖给不同的人。 # 这里是 crate 的版本。 -version = "0.64.0" +version = "0.64.1" [workspace.dependencies] # ── 内部 crate ─────────────────────────────────────────── diff --git a/crates/tw-api/msg-codes.txt b/crates/tw-api/msg-codes.txt index 2513b074..7c6dbac9 100644 --- a/crates/tw-api/msg-codes.txt +++ b/crates/tw-api/msg-codes.txt @@ -367,7 +367,9 @@ gw.probe.connect gw.probe.key_rejected gw.probe.request_failed gw.probe.timeout -gw.request.body_too_large +gw.request.body_declared_too_large +gw.request.body_over_limit +gw.request.body_unreadable gw.route.all_selected_disabled gw.route.denied gw.route.failed diff --git a/crates/tw-gateway/src/error.rs b/crates/tw-gateway/src/error.rs index 0cafb853..aa2af8d9 100644 --- a/crates/tw-gateway/src/error.rs +++ b/crates/tw-gateway/src/error.rs @@ -44,6 +44,9 @@ pub enum Source { /// 也没空出来。**和上游限流一样回 429,另外带 `Retry-After`**:请求本身没问题,过一会儿 /// 再来就能发出去 —— 客户端该退避再试,不是放弃。对外的词表里算 `rate_limited` Busy, + /// 请求体超过网关的上限(见 `server::intake`)。**413,不是 400**:Anthropic 的格式里 + /// 它是 `request_too_large`,和 Anthropic 自己嫌请求太大时一样。对外的词表里算 `request` + TooLarge, } /// 上游都满着时告诉客户端过几秒再来(`Retry-After`)。 @@ -62,7 +65,7 @@ impl Source { Source::Request => "request", Source::RateLimited | Source::Busy => "rate_limited", Source::Denied => "denied", - Source::NotSupported => "request", + Source::NotSupported | Source::TooLarge => "request", } } fn status(&self) -> StatusCode { @@ -75,6 +78,7 @@ impl Source { Source::RateLimited | Source::Busy => StatusCode::TOO_MANY_REQUESTS, Source::Denied => StatusCode::FORBIDDEN, Source::NotSupported => StatusCode::NOT_IMPLEMENTED, + Source::TooLarge => StatusCode::PAYLOAD_TOO_LARGE, } } } @@ -173,6 +177,9 @@ impl GatewayError { pub fn request(detail: Msg) -> Self { Self::new(Source::Request, detail) } + pub fn too_large(detail: Msg) -> Self { + Self::new(Source::TooLarge, detail) + } } impl GatewayError { diff --git a/crates/tw-gateway/src/forward.rs b/crates/tw-gateway/src/forward.rs index ad534a70..1af97948 100644 --- a/crates/tw-gateway/src/forward.rs +++ b/crates/tw-gateway/src/forward.rs @@ -140,17 +140,6 @@ pub fn map_reqwest_error(e: reqwest::Error) -> GatewayError { } } -/// 请求体原样转发,只在这里做一次大小检查。 -pub fn check_body_size(body: &Bytes, max: usize) -> Result<(), GatewayError> { - if body.len() > max { - return Err(GatewayError::request(msg!( - "gw.request.body_too_large", size = body.len(), max = max => - "The request body is {size} bytes, over the {max}-byte limit." - ))); - } - Ok(()) -} - #[cfg(test)] mod tests { use super::*; @@ -283,11 +272,4 @@ mod tests { "/v1beta/models/small:streamGenerateContent" ); } - - #[test] - fn body_size_limit_reports_both_numbers() { - let e = check_body_size(&Bytes::from(vec![0u8; 10]), 5).unwrap_err(); - assert!(e.message().contains("10") && e.message().contains('5')); - assert!(check_body_size(&Bytes::from(vec![0u8; 5]), 5).is_ok()); - } } diff --git a/crates/tw-gateway/src/server.rs b/crates/tw-gateway/src/server.rs index e161fb79..645a658e 100644 --- a/crates/tw-gateway/src/server.rs +++ b/crates/tw-gateway/src/server.rs @@ -7,7 +7,6 @@ use axum::extract::{OriginalUri, RawQuery, State}; use axum::http::HeaderMap; use axum::response::{IntoResponse, Response}; use axum::routing::{any, get}; -use bytes::Bytes; use crate::error::GatewayError; use crate::health::Health; @@ -16,6 +15,7 @@ use listing::{get_model, list_models}; pub(crate) use pipeline::stream_fault; use tw_types::msg; +mod intake; mod listing; mod pipeline; mod upgrade; @@ -98,11 +98,11 @@ async fn passthrough( method: axum::http::Method, OriginalUri(uri): OriginalUri, RawQuery(query): RawQuery, - // **必须排在 `body` 前面。**提取器按顺序跑,而 `Bytes` 会把体吃掉 - // —— 一次升级要的是那条连接本身,体被读走之后就没得升了 + // **必须排在 `body` 前面。**`body` 拿走的是整个请求 crate::ws::MaybeUpgrade(upgrade): crate::ws::MaybeUpgrade, headers: HeaderMap, - body: Bytes, + // **这里还没读。**过完来源和密钥检查才读(见 `intake`) + body: axum::body::Body, ) -> Result { let started = std::time::Instant::now(); // 从这一刻起它就算「在服务中」。**排队等并发名额的也算** —— 那时客户 @@ -135,9 +135,7 @@ async fn passthrough( .find(|c| c.name == client_name) .map(|c| tw_secret::mask_secret(&c.key)), }; - // WebSocket 升级。**在鉴权之后、解体之前分叉** —— 鉴权 - // 在前是因为一个不该连过来的地址不该有机会升级;解体之前是因为 - // 升级要的是那条连接,而 `Bytes` 会把它读干净。 + // WebSocket 升级。**在鉴权之后分叉** —— 一个不该连过来的地址不该有机会升级。 if let Some(ws) = upgrade.filter(|_| crate::ws::is_upgrade(&headers)) { return upgrade::ws_upgrade( state, @@ -189,6 +187,10 @@ async fn passthrough( // 交给管线里每一条 `return Err` 各自去报的话,漏掉一条的后果不是没报, // 而是被 Drop 报成「客户端取消」—— 一次策略拒绝会记到客户端头上。 let mut ending: Option = None; + // **到这里才读体**:来源、密钥、方法都过了。超了上限的还没有开始事件,不记一行 + let body = intake::read(&headers, body, intake::MAX_BODY) + .await + .map_err(|e| e.in_dialect(dialect))?; let req = pipeline::Inbound { uri, query, diff --git a/crates/tw-gateway/src/server/intake.rs b/crates/tw-gateway/src/server/intake.rs new file mode 100644 index 00000000..857f2f16 --- /dev/null +++ b/crates/tw-gateway/src/server/intake.rs @@ -0,0 +1,107 @@ +//! 读请求体。**整个网关只有这一道大小上限。** +//! +//! 体在处理函数里、过完来源和密钥检查之后才读(见 `passthrough`),不交给 axum 的 +//! `Bytes` 提取器:提取器在处理函数之前就把整个体读进内存,一个不在放行网段里、 +//! 没带密钥的连接也能先塞满上限再被拒;它自带的 2 MB 默认上限还会把贴了几张截图的 +//! 对话拒成 413,连压缩请求一起,会话就再也压不回来了(Core#295)。 + +use axum::body::Body; +use axum::http::{HeaderMap, header}; +use bytes::Bytes; +use http_body_util::{BodyExt, LengthLimitError, Limited}; + +use crate::error::GatewayError; +use tw_types::msg; + +/// 256 MiB。大到能装下几张 4K 图的 base64(膨胀 33%),小到失控的 +/// 客户端打不爆内存。 +pub(super) const MAX_BODY: usize = 256 * 1024 * 1024; + +/// 读完请求体,最多 `max` 字节。 +/// +/// **`Content-Length` 只用来提前回绝**:声明的已经超了,不读就拒,老实的客户端不用先 +/// 传完几百兆再挨一个 413。真正卡住的是读的时候按实际收到的字节计数 —— 声明的长度 +/// 是对方自己写的,所以报错里只说「声明的」。 +pub(super) async fn read( + headers: &HeaderMap, + body: Body, + max: usize, +) -> Result { + let declared = headers + .get(header::CONTENT_LENGTH) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()); + if let Some(size) = declared.filter(|&n| n > max as u64) { + return Err(GatewayError::too_large(msg!( + "gw.request.body_declared_too_large", size = size, max = max => + "The request declares a {size}-byte body, over the {max}-byte limit." + ))); + } + match Limited::new(body, max).collect().await { + Ok(c) => Ok(c.to_bytes()), + Err(e) if e.downcast_ref::().is_some() => { + Err(GatewayError::too_large(msg!( + "gw.request.body_over_limit", max = max => + "The request body is over the {max}-byte limit." + ))) + } + // 客户端传到一半断了。它多半已经听不到这句了,但这一行得有个结局 + Err(e) => Err(GatewayError::request(msg!( + "gw.request.body_unreadable", detail = e => + "The request body could not be read: {detail}" + ))), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::http::HeaderValue; + + fn chunked(parts: &[&'static [u8]]) -> Body { + let parts: Vec> = + parts.iter().map(|p| Ok(Bytes::from_static(p))).collect(); + Body::from_stream(futures::stream::iter(parts)) + } + + #[tokio::test] + async fn a_body_at_the_limit_is_read_whole() { + let b = read(&HeaderMap::new(), chunked(&[b"12", b"345"]), 5) + .await + .unwrap(); + assert_eq!(&b[..], b"12345"); + } + + #[tokio::test] + async fn a_body_without_a_length_stops_at_the_limit() { + let e = read(&HeaderMap::new(), chunked(&[b"123", b"456"]), 5) + .await + .unwrap_err(); + assert_eq!(e.detail.code, "gw.request.body_over_limit"); + assert_eq!(e.source, crate::error::Source::TooLarge); + } + + #[tokio::test] + async fn a_declared_length_over_the_limit_is_refused_unread() { + let mut h = HeaderMap::new(); + h.insert(header::CONTENT_LENGTH, HeaderValue::from_static("6")); + // 体是一条永远不结束的流:读了就挂住,拒得掉说明根本没读 + let e = read( + &h, + Body::from_stream(futures::stream::pending::>()), + 5, + ) + .await + .unwrap_err(); + assert_eq!(e.detail.code, "gw.request.body_declared_too_large"); + assert_eq!(e.detail.args["size"], "6"); + } + + #[tokio::test] + async fn a_declared_length_that_understates_the_body_does_not_lift_the_limit() { + let mut h = HeaderMap::new(); + h.insert(header::CONTENT_LENGTH, HeaderValue::from_static("1")); + let e = read(&h, chunked(&[b"123", b"456"]), 5).await.unwrap_err(); + assert_eq!(e.detail.code, "gw.request.body_over_limit"); + } +} diff --git a/crates/tw-gateway/src/server/pipeline.rs b/crates/tw-gateway/src/server/pipeline.rs index 243dccc4..07f56865 100644 --- a/crates/tw-gateway/src/server/pipeline.rs +++ b/crates/tw-gateway/src/server/pipeline.rs @@ -16,7 +16,6 @@ use bytes::Bytes; use super::{Choice, Sender, now_ms}; use crate::error::GatewayError; -use crate::forward; use crate::state::{AppState, Runtime}; use tw_types::msg; @@ -29,10 +28,6 @@ mod slow; pub(crate) use hop::stream_fault; -/// 256 MiB。大到能装下几张 4K 图的 base64(膨胀 33%),小到失控的 -/// 客户端打不爆内存。 -const MAX_BODY: usize = 256 * 1024 * 1024; - /// 进管线时就定了的东西:身份识别之后,每一步都只读不改。 pub(super) struct Inbound { pub(super) uri: axum::http::Uri, @@ -76,8 +71,6 @@ pub(super) async fn pipeline( live: crate::live::Pass, ending: &mut Option, ) -> Result { - forward::check_body_size(&req.body, MAX_BODY)?; - let intent = match probe(&state, &rt, &req) { Probe::Answered(resp) => return Ok(resp), Probe::Intent(intent) => intent, diff --git a/crates/tw-gateway/src/wire.rs b/crates/tw-gateway/src/wire.rs index 897be607..51c2fb9c 100644 --- a/crates/tw-gateway/src/wire.rs +++ b/crates/tw-gateway/src/wire.rs @@ -13,7 +13,7 @@ impl From for FailureSource { Source::Request => Self::Request, Source::RateLimited | Source::Busy => Self::RateLimited, Source::Denied => Self::Denied, - Source::NotSupported => Self::Request, + Source::NotSupported | Source::TooLarge => Self::Request, } } } diff --git a/crates/tw-gateway/tests/body_size.rs b/crates/tw-gateway/tests/body_size.rs new file mode 100644 index 00000000..07d7cceb --- /dev/null +++ b/crates/tw-gateway/tests/body_size.rs @@ -0,0 +1,136 @@ +//! 请求体的大小上限(Core#295):网关只有自己那一道 256 MiB,体在鉴权之后才读。 +//! +//! 原来体由 axum 的 `Bytes` 提取器读,它自带 2 MB 的默认上限 —— 贴了几张截图的对话 +//! 一过 2 MiB 就被拒成 413,到不了网关自己的检查;而且体在来源、密钥检查之前就被 +//! 整个读进内存。 + +use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use axum::Router; +use axum::extract::{DefaultBodyLimit, State}; +use axum::routing::post; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tw_config::{Client, Config, Listen, Provider}; + +/// 假上游。**它自己也得关掉 axum 的 2 MB 上限**,不然大请求过了网关、倒在这里 +async fn start_upstream() -> (SocketAddr, Arc>) { + let got = Arc::new(Mutex::new(0usize)); + let app = Router::new() + .route( + "/v1/messages", + post( + |State(got): State>>, body: bytes::Bytes| async move { + *got.lock().unwrap() = body.len(); + axum::response::Response::builder() + .header("content-type", "application/json") + .body(axum::body::Body::from( + r#"{"type":"message","id":"msg_01"}"#, + )) + .unwrap() + }, + ), + ) + .layer(DefaultBodyLimit::disable()) + .with_state(got.clone()); + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = l.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(l, app).await.unwrap() }); + (addr, got) +} + +async fn start_gateway(upstream: SocketAddr) -> SocketAddr { + let cfg = Config { + version: 1, + listen: Listen::default(), + clients: vec![Client { + name: "claude-code".into(), + key: "tw-testkey".into(), + ..Default::default() + }], + providers: vec![Provider { + name: "mock".into(), + base_url: format!("http://{upstream}"), + key: Some("sk-upstream".into()), + protocol: Some(tw_config::Protocol::Anthropic), + ..Default::default() + }], + ..Default::default() + }; + let state = tw_gateway::AppState::new(cfg).unwrap(); + tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap() +} + +/// 一个 `size` 字节、合法的 Anthropic 请求:一段很长的用户消息 +fn body_of(size: usize) -> String { + let head = + r#"{"model":"claude-sonnet-4-5","max_tokens":16,"messages":[{"role":"user","content":""#; + let tail = r#""}]}"#; + format!("{head}{}{tail}", "x".repeat(size - head.len() - tail.len())) +} + +#[tokio::test] +async fn a_request_over_axums_default_two_mib_reaches_the_upstream() { + let (up, got) = start_upstream().await; + let gw = start_gateway(up).await; + let body = body_of(3 * 1024 * 1024); + + let resp = reqwest::Client::new() + .post(format!("http://{gw}/v1/messages")) + .header("x-api-key", "tw-testkey") + .header("anthropic-version", "2023-06-01") + .header("content-type", "application/json") + .body(body.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200, "{}", resp.text().await.unwrap()); + assert_eq!(*got.lock().unwrap(), body.len()); +} + +/// 只发请求头、声明一个 `length` 字节的体、一个字节都不发,读回状态行和整个响应。 +/// +/// **回得来就说明网关没在等体**:体要是先读,这里会一直挂到超时 +async fn headers_only(gw: SocketAddr, key: &str, length: u64) -> (u16, String) { + let mut s = tokio::net::TcpStream::connect(gw).await.unwrap(); + let req = format!( + "POST /v1/messages HTTP/1.1\r\nhost: {gw}\r\nx-api-key: {key}\r\n\ + anthropic-version: 2023-06-01\r\ncontent-type: application/json\r\n\ + content-length: {length}\r\nconnection: close\r\n\r\n" + ); + s.write_all(req.as_bytes()).await.unwrap(); + let mut out = Vec::new(); + tokio::time::timeout(Duration::from_secs(5), s.read_to_end(&mut out)) + .await + .expect("网关在等请求体,没有先回答") + .unwrap(); + let text = String::from_utf8_lossy(&out).into_owned(); + let status = text + .split(' ') + .nth(1) + .and_then(|c| c.parse().ok()) + .unwrap_or_else(|| panic!("不是 HTTP 响应:{text}")); + (status, text) +} + +#[tokio::test] +async fn a_request_with_a_wrong_key_is_refused_before_its_body_is_read() { + let (up, _) = start_upstream().await; + let gw = start_gateway(up).await; + let (status, _) = headers_only(gw, "tw-wrong", 100 * 1024 * 1024).await; + assert_eq!(status, 401); +} + +#[tokio::test] +async fn a_declared_length_over_the_limit_is_refused_unread_in_the_clients_format() { + let (up, got) = start_upstream().await; + let gw = start_gateway(up).await; + let (status, text) = headers_only(gw, "tw-testkey", 300 * 1024 * 1024).await; + assert_eq!(status, 413); + assert!(text.contains("request_too_large"), "{text}"); + assert!(text.contains("x-thinkwatch-error: request"), "{text}"); + assert_eq!(*got.lock().unwrap(), 0, "超限的请求不该发到上游"); +} diff --git a/release-notes/0.64.1.md b/release-notes/0.64.1.md new file mode 100644 index 00000000..ecf58d99 --- /dev/null +++ b/release-notes/0.64.1.md @@ -0,0 +1,14 @@ +This release lifts a 2 MiB limit on request bodies that the gateway never meant to have. A conversation holding a few inline screenshots passed 2 MiB and was refused from then on, compaction requests included, so the session could not be compacted back under the limit (#295). + +**Upgrade notes** + +- The control-plane protocol version (`CONTROL_API_VERSION`) stays at 40: ThinkWatch Lite 2026.10.7 connects to 0.64.1, and a server can be upgraded on its own. +- The request store's schema is unchanged: upgrading from 0.64.0 keeps the request history. +- The configuration format is unchanged. +- Message codes, compared with 0.64.0: + - New: `gw.request.body_declared_too_large`, `gw.request.body_over_limit`, `gw.request.body_unreadable`. + - Removed: `gw.request.body_too_large`, replaced by the first two: one for a declared length over the limit, which names the length the client declared, and one for a body found over the limit while reading, which has no size to name. + +**Request size.** The gateway's only limit on a request body is now its own 256 MiB. The body used to be read by the web framework before the gateway saw the request, under the framework's default 2 MiB limit: anything larger got a plain-text `413 Failed to buffer the request body`, was not in the client's error format, and left no record. + +The body is now read only after the source-address check, the key check and the POST check pass, so a connection that is not allowed in can no longer make the gateway hold its body in memory first. A request whose `Content-Length` is already over the limit is refused without reading the body. The limit is enforced on the bytes actually received, so a `Content-Length` that understates the body does not get around it. An oversized request is answered with 413 in the client's own error format (`request_too_large` for Anthropic clients), and `x-thinkwatch-error: request`.