diff --git a/backend/crates/providers/openai/tests/transport/tls.rs b/backend/crates/providers/openai/tests/transport/tls.rs index fd7da18a0..baba51dd2 100644 --- a/backend/crates/providers/openai/tests/transport/tls.rs +++ b/backend/crates/providers/openai/tests/transport/tls.rs @@ -33,31 +33,23 @@ fn http_client_hellos_should_match_official_tls_transports() { .enable_all() .build() .unwrap(); - let hello = runtime.block_on(async { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let url = format!( - "https://localhost:{}/", - listener.local_addr().unwrap().port() - ); + runtime.block_on(async { let client = provider_openai::transport::build_reqwest_client().unwrap(); - let (hello, response) = timeout(Duration::from_secs(10), async { - tokio::join!(read_client_hello(listener), client.get(url).send()) - }) - .await - .expect("HTTP ClientHello within timeout"); - assert!( - response.is_err(), - "capture endpoint rejects TLS after ClientHello" - ); - hello + let mut hello = capture_http_hello(client).await; + if case == "custom-ca" { + hello.extensions.sort_unstable(); + assert_eq!(hello, official_http_rustls_hello()); + } else { + #[cfg(target_os = "linux")] + let expected = official_linux_http_hello(); + // 官方无自定义 CA 时使用 reqwest 默认的原生 TLS；系统后端的握手随操作系统而变 + #[cfg(not(target_os = "linux"))] + let expected = + capture_http_hello(reqwest::Client::builder().no_proxy().build().unwrap()) + .await; + assert_eq!(hello, expected); + } }); - if case == "custom-ca" { - let mut normalized = hello; - normalized.extensions.sort_unstable(); - assert_eq!(normalized, official_http_rustls_hello()); - } else { - assert_eq!(hello, official_http_hello()); - } println!("\n{CASE_COMPLETED}"); return; } @@ -99,6 +91,24 @@ fn http_client_hellos_should_match_official_tls_transports() { } } +async fn capture_http_hello(client: reqwest::Client) -> ClientHello { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!( + "https://localhost:{}/", + listener.local_addr().unwrap().port() + ); + let (hello, response) = timeout(Duration::from_secs(10), async { + tokio::join!(read_client_hello(listener), client.get(url).send()) + }) + .await + .expect("HTTP ClientHello within timeout"); + assert!( + response.is_err(), + "capture endpoint rejects TLS after ClientHello" + ); + hello +} + #[tokio::test] async fn websocket_client_hello_should_match_official_rustls_transport() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -124,7 +134,8 @@ async fn websocket_client_hello_should_match_official_rustls_transport() { assert_eq!(normalized, official_websocket_hello()); } -fn official_http_hello() -> ClientHello { +#[cfg(target_os = "linux")] +fn official_linux_http_hello() -> ClientHello { ClientHello { legacy_version: 771, supported_versions: vec![772, 771], @@ -203,16 +214,27 @@ async fn read_client_hello(listener: TcpListener) -> ClientHello { assert!(values.insert(kind, take_vector(&mut extensions)).is_none()); } - let mut groups = values[&10].clone(); - let groups = u16_values(take_vector(&mut groups)); - let mut signatures = values[&13].clone(); - let signature_algorithms = u16_values(take_vector(&mut signatures)); - let mut shares = values[&51].clone(); - let mut shares = take_vector(&mut shares); + // 原生后端可能只发送 TLS 1.2 扩展；扩展是否存在仍由完整的类型列表校验 + let u16_list = |kind| { + values + .get(&kind) + .cloned() + .map(|mut data| u16_values(take_vector(&mut data))) + .unwrap_or_default() + }; + let byte_list = |kind| { + values + .get(&kind) + .map(|data| data.slice(1..)) + .unwrap_or_default() + }; let mut key_shares = Vec::new(); - while shares.has_remaining() { - let group = shares.get_u16(); - key_shares.push((group, take_vector(&mut shares).len())); + if let Some(mut data) = values.get(&51).cloned() { + let mut shares = take_vector(&mut data); + while shares.has_remaining() { + let group = shares.get_u16(); + key_shares.push((group, take_vector(&mut shares).len())); + } } let mut alpn = Vec::new(); if let Some(protocols) = values.get(&16) { @@ -225,13 +247,13 @@ async fn read_client_hello(listener: TcpListener) -> ClientHello { } ClientHello { legacy_version, - supported_versions: u16_values(values[&43].slice(1..)), - point_formats: values[&11].slice(1..).to_vec(), - psk_modes: values[&45].slice(1..).to_vec(), + supported_versions: u16_values(byte_list(43)), + point_formats: byte_list(11).to_vec(), + psk_modes: byte_list(45).to_vec(), cipher_suites, extensions: extension_order, - groups, - signature_algorithms, + groups: u16_list(10), + signature_algorithms: u16_list(13), key_shares, alpn, } diff --git a/backend/crates/providers/openai/tests/transport/websocket/mod.rs b/backend/crates/providers/openai/tests/transport/websocket/mod.rs index 304c4cfe2..1727210f9 100644 --- a/backend/crates/providers/openai/tests/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/tests/transport/websocket/mod.rs @@ -1451,18 +1451,16 @@ async fn websocket_execute_response_create_request_should_return_error_terminal_ assert!(response.body.contains("No tool output found")); } -#[tokio::test(start_paused = true)] +#[tokio::test] async fn codex_backend_client_should_timeout_when_upstream_is_silent() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let server = tokio::spawn(async move { - let mut websockets = Vec::new(); - for _ in 0..2 { - let (stream, _) = listener.accept().await.unwrap(); - let mut websocket = accept_codex_test_websocket(stream).await; - let _message = websocket.next().await.unwrap().unwrap(); - websockets.push(websocket); - } + let (stream, _) = listener.accept().await.unwrap(); + let mut websocket = accept_codex_test_websocket(stream).await; + let _message = websocket.next().await.unwrap().unwrap(); + // 握手和首条请求使用实时时钟，随后让虚拟时钟推进到客户端的空闲期限 + tokio::time::pause(); futures::future::pending::<()>().await; }); let backend = CodexBackendClient::new( @@ -1700,12 +1698,13 @@ async fn codex_backend_client_stream_should_reject_binary_websocket_event() { ); } -#[tokio::test(start_paused = true)] +#[tokio::test] async fn codex_backend_client_stream_should_keep_socket_after_structural_activity() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let accepted_connections = Arc::new(AtomicUsize::new(0)); let accepted_connections_for_server = Arc::clone(&accepted_connections); + let (output_tx, output_rx) = tokio::sync::oneshot::channel(); let server = tokio::spawn(async move { let (first_stream, _) = listener.accept().await.unwrap(); accepted_connections_for_server.fetch_add(1, Ordering::SeqCst); @@ -1726,7 +1725,7 @@ async fn codex_backend_client_stream_should_keep_socket_after_structural_activit .await .unwrap(); - tokio::time::sleep(Duration::from_secs(30)).await; + output_rx.await.unwrap(); first_websocket .send(Message::Text( json!({ @@ -1768,7 +1767,13 @@ async fn codex_backend_client_stream_should_keep_socket_after_structural_activit .expect("structural activity should keep the websocket stream open"); let decision = response.websocket_pool_decision; let mut stream = response.body; - let mut body = String::new(); + let first = stream.next().await.expect("structural frame").unwrap(); + let mut body = std::str::from_utf8(&first).unwrap().to_owned(); + assert!(body.contains("response.created")); + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(30)).await; + tokio::time::resume(); + output_tx.send(()).unwrap(); while let Some(chunk) = stream.next().await { let chunk = chunk.unwrap_or_else(|error| { panic!("delayed websocket stream chunk should be valid: {error}") @@ -2015,6 +2020,12 @@ async fn codex_backend_client_stream_should_cancel_while_inbound_is_backpressure | Some(Err(tungstenite::Error::Protocol( tungstenite::error::ProtocolError::ResetWithoutClosingHandshake, ))) => return, + // 取消时若 TCP 仍有未读帧，对端可直接收到连接重置 + Some(Err(tungstenite::Error::Io(error))) + if error.kind() == std::io::ErrorKind::ConnectionReset => + { + return; + } Some(Ok(_)) => continue, Some(Err(error)) => panic!("unexpected websocket cleanup error: {error}"), } @@ -2046,10 +2057,11 @@ async fn codex_backend_client_stream_should_cancel_while_inbound_is_backpressure server.await.unwrap(); } -#[tokio::test(start_paused = true)] +#[tokio::test] async fn codex_backend_client_stream_should_wait_for_terminal_after_active_websocket_gap() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); + let (finish_tx, finish_rx) = tokio::sync::oneshot::channel(); let server = tokio::spawn(async move { let (stream, _) = listener.accept().await.unwrap(); let mut websocket = accept_codex_test_websocket(stream).await; @@ -2065,7 +2077,7 @@ async fn codex_backend_client_stream_should_wait_for_terminal_after_active_webso )) .await .unwrap(); - tokio::time::sleep(Duration::from_secs(30)).await; + finish_rx.await.unwrap(); websocket .send(Message::Text( completed_websocket_response("resp_after_active_gap", 3, 1).into(), @@ -2098,8 +2110,10 @@ async fn codex_backend_client_stream_should_wait_for_terminal_after_active_webso .expect("stream should yield partial frame") .expect("partial frame should be valid"); assert!(std::str::from_utf8(&first).unwrap().contains("partial")); + tokio::time::pause(); tokio::time::advance(Duration::from_secs(30)).await; - tokio::task::yield_now().await; + tokio::time::resume(); + finish_tx.send(()).unwrap(); let terminal = response .body .next()