diff --git a/.github/workflows/reality-interop.yml b/.github/workflows/reality-interop.yml index 9c9a971a..696428ff 100644 --- a/.github/workflows/reality-interop.yml +++ b/.github/workflows/reality-interop.yml @@ -16,8 +16,8 @@ jobs: include: - name: v26.10.10 revision: 701af60772cda123492f86e383e1fdba066614a9 - - name: main-2026-10-14 - revision: 0888beb38a9ba45440557175e71ce21cf8ba3b13 + - name: v26.10.17 + revision: 92ccc939aaf4f8965f369cf0b142e005aeaa11d5 name: Xray ${{ matrix.name }} env: CARGO_BUILD_JOBS: 1 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index d84196fa..935d1a1a 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -13,9 +13,9 @@ env: SING_BOX_SHA256: 12cb2816b52febb356f6a885b740cc8758c3f30b8ae0ca8edba80f0d2d35343f MIHOMO_VERSION: 1.19.31 MIHOMO_SHA256: 04cf9f09671704f839ddbee2e93069dc831a4123a75281e725d1d96ab9ac1afc - XRAY_VERSION: 26.3.27 - XRAY_SHA256: 23cd9af937744d97776ee35ecad4972cf4b2109d1e0fe6be9930467608f7c8ae - XRAY_COMMIT: 701af60772cda123492f86e383e1fdba066614a9 + XRAY_VERSION: 26.10.17 + XRAY_SHA256: ecdd54eccd24c4c21a8f00968e7dbf7f5996da92383d2321170b560dc4a95158 + XRAY_COMMIT: 92ccc939aaf4f8965f369cf0b142e005aeaa11d5 MIHOMO_ALPHA_COMMIT: 94b13aed49b8dc41973e4bb3f721c5fa2c5c5c1f NAIVE_CADDY_VERSION: 2.11.7-naive NAIVE_CADDY_SHA256: ac97057c925bb098eb575e7b0cb8bd547d4c5f845b09a3dedfbe34444aa7ae88 @@ -45,7 +45,7 @@ jobs: env: GOMAXPROCS: 1 WG_SING_BOX_COMMIT: 7054cac5124465ffdb145870d7c01885566711b1 - WG_XRAY_COMMIT: 0888beb38a9ba45440557175e71ce21cf8ba3b13 + WG_XRAY_COMMIT: 92ccc939aaf4f8965f369cf0b142e005aeaa11d5 steps: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@stable diff --git a/.github/workflows/vless-encryption-interop.yml b/.github/workflows/vless-encryption-interop.yml index ebd62248..a36100ec 100644 --- a/.github/workflows/vless-encryption-interop.yml +++ b/.github/workflows/vless-encryption-interop.yml @@ -25,7 +25,7 @@ jobs: - uses: actions/checkout@v4 with: repository: XTLS/Xray-core - ref: 0f3d884786f9ec73d3be8899cb6b9e1e6dd3c41e + ref: 92ccc939aaf4f8965f369cf0b142e005aeaa11d5 path: peers/xray - uses: actions/checkout@v4 with: @@ -69,6 +69,7 @@ jobs: env: XRAY_BINARY: ${{ runner.temp }}/xray MIHOMO_BINARY: ${{ runner.temp }}/mihomo + SHOES_TEST_SING_BOX_BIN: ${{ runner.temp }}/sing-box run: cargo test --locked --features optional-peer-tests --test vless_encryption_interop -- --test-threads=1 - name: Cross-language cryptography vectors run: | diff --git a/README.md b/README.md index c7bc8ef5..8833870c 100644 --- a/README.md +++ b/README.md @@ -557,14 +557,14 @@ covering all three modes and HTTP versions. `type: splithttp` is an alias for transport setup (or the shorter configured setup timeout). H1 keep-alive idle periods and subsequent partial headers have no deadline. For exposed listeners, set `global_limits.max_connections` and `global_limits.max_connections_per_ip`. -- Xray 26.10.10's H1 packet-up uploader can lose pipelined requests when Go GC - discards pooled connections with unread responses. Sustained uploads to slow - destinations can therefore fail despite bounded server backpressure. Prefer - H2/H3 or a streaming mode for these clients; disabling GC is not a production fix. +- Xray's H1 packet-up uploader (observed on 26.10.10) can lose pipelined requests + when Go GC discards pooled connections with unread responses. Sustained uploads + to slow destinations can therefore fail despite bounded server backpressure. + Prefer H2/H3 or a streaming mode for these clients; disabling GC is not a production fix. - Advanced placement/padding obfuscation, custom session IDs, separate downloads, - full XMUX policy, and Vision over XHTTP are not supported. Unsupported options - are rejected rather than silently ignored. Public CDN compatibility is not implied. -- REALITY XHTTP interoperates with Xray 26.10.10 in both directions over H2, + full XMUX policy, and legacy Vision over XHTTP are not supported. Unsupported + options are rejected rather than silently ignored. Public CDN compatibility is not implied. +- REALITY XHTTP interoperates with Xray 26.10.17 in both directions over H2, including all three modes and `auto`. REALITY prefers X25519MLKEM768 with X25519 fallback according to the cover target. Optional `mldsa65_seed` and `mldsa65_verify` apply to the outer REALITY configuration, independently of diff --git a/src/vless/encryption/handshake.rs b/src/vless/encryption/handshake.rs index 7c8594ac..6e9146eb 100644 --- a/src/vless/encryption/handshake.rs +++ b/src/vless/encryption/handshake.rs @@ -155,6 +155,7 @@ impl ClientEncryption { used, }, )?; + log::debug!("VLESS encryption client: resumed handshake"); return Ok(stream); } let mlkem = kem::DecapsulationKey::generate(&kem::ML_KEM_768) @@ -276,6 +277,7 @@ impl ClientEncryption { .map_err(|_| { io::Error::new(io::ErrorKind::TimedOut, "VLESS fresh handshake timed out") })??; + log::debug!("VLESS encryption client: fresh handshake"); Ok(EncryptionStream::new(io, true, keys, initial)) } @@ -319,6 +321,14 @@ impl ServerEncryption { if let Some(random) = random_prefix { stream.set_server_random(random); } + log::debug!( + "VLESS encryption server: {} handshake", + if random_prefix.is_some() { + "resumed" + } else { + "fresh" + } + ); Ok(stream) } @@ -345,6 +355,7 @@ impl ServerEncryption { }; let length = u16::from_be_bytes(plain.try_into().unwrap()) as usize; let cipher = nfs_aead.cipher; + log::debug!("VLESS encryption server cipher: {cipher:?}"); let random_mode = self.mode == Mode::Random; if length == 32 { let mut encrypted_ticket = [0; 32]; diff --git a/src/vless/encryption/mod.rs b/src/vless/encryption/mod.rs index 934a7c24..4605db5c 100644 --- a/src/vless/encryption/mod.rs +++ b/src/vless/encryption/mod.rs @@ -8,7 +8,7 @@ mod session; mod stream; #[cfg(test)] -mod tests; +pub(crate) mod tests; use std::{fmt, io, sync::Arc}; diff --git a/src/vless/encryption/stream.rs b/src/vless/encryption/stream.rs index e4d6ff0f..a87e2758 100644 --- a/src/vless/encryption/stream.rs +++ b/src/vless/encryption/stream.rs @@ -131,7 +131,7 @@ pub(super) struct StreamKeys { impl EncryptionStream { #[cfg(test)] - pub(super) fn is_resumed(&self) -> bool { + pub(crate) fn is_resumed(&self) -> bool { self.resume.is_some() } pub(super) fn new(inner: S, client: bool, keys: StreamKeys, initial: InitialRead) -> Self { diff --git a/src/vless/encryption/stream/tests.rs b/src/vless/encryption/stream/tests.rs index 039c711a..3eca9ac8 100644 --- a/src/vless/encryption/stream/tests.rs +++ b/src/vless/encryption/stream/tests.rs @@ -7,6 +7,54 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; mod liveness; mod resumption; +#[tokio::test] +async fn maximum_native_record_keeps_direct_command_in_second_vision_read_and_raw_tail() { + use crate::vless::vision_stream::VisionStream; + for masked in [false, true] { + for cipher in [Cipher::Aes256Gcm, Cipher::ChaCha20Poly1305] { + let first = vec![41; 9000]; + let second = vec![42; 7596]; + let mut plain = vec![0, 0]; + plain.extend_from_slice(&[7; 16]); + plain.push(0); + plain.extend_from_slice(&(first.len() as u16).to_be_bytes()); + plain.extend_from_slice(&[0, 0]); + plain.extend_from_slice(&first); + plain.push(2); + plain.extend_from_slice(&(second.len() as u16).to_be_bytes()); + plain.extend_from_slice(&[0, 0]); + plain.extend_from_slice(&second); + assert_eq!(plain.len(), 16624); + let mut peer = Aead::new(b"context", &[7; 96], cipher).unwrap(); + let mut wire = peer.seal_record(&plain, &[7; 96]).unwrap(); + wire.extend_from_slice(b"\x17\x03\x03\x00\x04tail"); + if masked { + HeaderMask::new(&[7; 96], &[8; 16], 0) + .unwrap() + .apply_write(&mut wire) + .unwrap(); + } + let mut stream = stream( + Scripted { + input: wire.into(), + eof: true, + ..Default::default() + }, + true, + masked, + ); + stream.read_aead = Some(Aead::new(b"context", &[7; 96], cipher).unwrap()); + let mut vision = VisionStream::new_client(stream, [7; 16]).unwrap(); + let mut received = Vec::new(); + vision.read_to_end(&mut received).await.unwrap(); + assert_eq!( + received, + [first, second, b"\x17\x03\x03\x00\x04tail".to_vec()].concat() + ); + } + } +} + #[derive(Default)] struct Scripted { input: VecDeque, diff --git a/src/vless/encryption/tests.rs b/src/vless/encryption/tests.rs index b0d514b4..f941ae63 100644 --- a/src/vless/encryption/tests.rs +++ b/src/vless/encryption/tests.rs @@ -8,6 +8,7 @@ use super::{ }; mod lifecycle; +mod listeners; mod transports; pub(super) fn fixtures() -> Value { @@ -164,7 +165,7 @@ pub(super) fn fixture_settings(mode: &str, resume: bool, keys: &[usize]) -> (Str ) } -pub(super) fn fixture_pair( +pub(crate) fn fixture_pair( mode: &str, resume: bool, keys: &[usize], diff --git a/src/vless/encryption/tests/listeners.rs b/src/vless/encryption/tests/listeners.rs new file mode 100644 index 00000000..cb303d6b --- /dev/null +++ b/src/vless/encryption/tests/listeners.rs @@ -0,0 +1,253 @@ +use std::{io, sync::Arc, time::Duration}; + +use serde_json::json; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + time::timeout, +}; + +use super::{ + fixture_settings, + transports::{Tasks, echo, request}, +}; +use crate::{ + async_stream::AsyncStream, config::ServerConfig, resolver::NativeResolver, + vless::encryption::ClientEncryption, +}; + +const LIMIT: Duration = Duration::from_secs(10); +const UUID: &str = "550e8400-e29b-41d4-a716-446655440000"; + +async fn port() -> u16 { + TcpListener::bind("0.0.0.0:0") + .await + .unwrap() + .local_addr() + .unwrap() + .port() +} + +async fn listen(config: serde_json::Value) -> Tasks { + let config: ServerConfig = serde_json::from_value(config).unwrap(); + Tasks( + crate::tcp::tcp_server::start_servers( + crate::config::Config::Server(config), + Arc::new(NativeResolver::new()), + ) + .await + .unwrap(), + ) +} + +async fn exchange(stream: &mut (impl AsyncStream + ?Sized), target: u16) -> io::Result<()> { + stream.write_all(&request(target, 1, false)).await?; + stream.write_all(b"scope").await?; + stream.flush().await?; + let mut reply = [0; 7]; + stream.read_exact(&mut reply).await?; + assert_eq!(&reply, b"\0\0scope"); + Ok(()) +} + +#[tokio::test] +async fn simultaneous_resumed_replays_across_bind_addresses_reach_destination_once() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::{sync::Barrier, task::JoinSet}; + + timeout(LIMIT, async { + let (encryption, decryption) = fixture_settings("random", true, &[32]); + let client = ClientEncryption::new(&encryption).unwrap(); + let port = port().await; + let mut listeners = listen(json!({ + "addresses":[format!("127.0.0.1:{port}"),format!("127.0.0.2:{port}")], + "protocol":{"type":"vless","user_id":UUID,"decryption":decryption} + })) + .await; + let target = TcpListener::bind("0.0.0.0:0").await.unwrap(); + let target_port = target.local_addr().unwrap().port(); + let accepted = Arc::new(AtomicUsize::new(0)); + let count = accepted.clone(); + let mut destination = Tasks(vec![tokio::spawn(async move { + let mut tasks = JoinSet::new(); + loop { + tokio::select! { + result = target.accept() => { + let (mut stream, _) = result.unwrap(); + count.fetch_add(1, Ordering::SeqCst); + tasks.spawn(async move { + let (mut r, mut w) = stream.split(); + let _ = tokio::io::copy(&mut r, &mut w).await; + }); + } + _ = tasks.join_next(), if !tasks.is_empty() => {} + } + } + })]); + let mut warm = client + .handshake(TcpStream::connect(("127.0.0.1", port)).await.unwrap()) + .await + .unwrap(); + exchange(&mut warm, target_port).await.unwrap(); + drop(warm); + assert_eq!(accepted.load(Ordering::SeqCst), 1); + + // Capture an unspent resumed flight, then race its identical bytes on both listeners. + let (a, mut b) = tokio::io::duplex(8192); + let mut resumed = client.handshake(a).await.unwrap(); + assert!(resumed.is_resumed()); + resumed + .write_all(&request(target_port, 1, false)) + .await + .unwrap(); + resumed.write_all(b"scope").await.unwrap(); + resumed.shutdown().await.unwrap(); + drop(resumed); + let mut flight = Vec::new(); + b.read_to_end(&mut flight).await.unwrap(); + let barrier = Arc::new(Barrier::new(16)); + let mut tasks = JoinSet::new(); + for i in 0..16 { + let barrier = barrier.clone(); + let flight = flight.clone(); + tasks.spawn(async move { + let mut stream = + TcpStream::connect((if i % 2 == 0 { "127.0.0.1" } else { "127.0.0.2" }, port)) + .await + .unwrap(); + barrier.wait().await; + stream.write_all(&flight).await.unwrap(); + stream.shutdown().await.unwrap(); + let mut response = Vec::new(); + // A rejected flight may reset TCP with unread application bytes. + // Wait for EOF/reset, not just the server's initial random prefix. + if let Err(error) = stream.read_to_end(&mut response).await { + assert_eq!(error.kind(), io::ErrorKind::ConnectionReset); + } + }); + } + while let Some(result) = tasks.join_next().await { + result.unwrap(); + } + // A final echo drains the accept queue before the duplicate-count assertion. + let mut sentinel = TcpStream::connect(("127.0.0.1", target_port)) + .await + .unwrap(); + sentinel.write_all(b"done").await.unwrap(); + sentinel.read_exact(&mut [0; 4]).await.unwrap(); + assert_eq!(accepted.load(Ordering::SeqCst), 3); + listeners.stop().await; + destination.stop().await; + }) + .await + .unwrap(); +} + +#[derive(Debug)] +struct Carrier; + +#[async_trait::async_trait] +impl crate::tcp::tcp_handler::TcpClientHandler for Carrier { + async fn setup_client_tcp_stream( + &self, + stream: Box, + _: crate::address::ResolvedLocation, + ) -> io::Result { + Ok(crate::tcp::tcp_handler::TcpClientSetupResult { + client_stream: stream, + early_data: None, + }) + } +} + +#[tokio::test] +async fn native_tickets_are_shared_across_bind_ips_but_isolated_by_sni_and_websocket_path() { + use crate::{ + address::{Address, NetLocation, ResolvedLocation}, + tcp::tcp_handler::TcpClientHandler, + tls_client_handler::TlsClientHandler, + websocket::WebsocketTcpClientHandler, + }; + timeout(LIMIT, async { + for tls in [false, true] { + let (encryption, decryption) = fixture_settings("native", true, &[32]); + let client = ClientEncryption::new(&encryption).unwrap(); + let cert = + rcgen::generate_simple_self_signed(vec!["a.local".into(), "b.local".into()]).unwrap(); + let protocol = json!({"type":"vless","user_id":UUID,"decryption":decryption}); + let outer = if tls { + let target = json!({"cert":cert.cert.pem(),"key":cert.signing_key.serialize_pem(),"protocol":protocol}); + json!({"type":"tls","tls_targets":{"a.local":target,"b.local":target}}) + } else { + json!({"type":"websocket","targets":[ + {"matching_path":"/a","protocol":protocol}, + {"matching_path":"/b","protocol":protocol} + ]}) + }; + let port = port().await; + let mut listeners = listen(json!({ + "addresses":[format!("127.0.0.1:{port}"),format!("127.0.0.2:{port}")],"protocol":outer + })) + .await; + let (target, mut destination) = echo().await; + let mut roots = rustls::RootCertStore::empty(); + roots.add(cert.cert.der().clone()).unwrap(); + let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider()); + let config = Arc::new( + rustls::ClientConfig::builder_with_provider(provider) + .with_safe_default_protocol_versions() + .unwrap() + .with_root_certificates(roots) + .with_no_client_auth(), + ); + for (scope, ip, resumed, rejected) in [ + ("a", 1, false, false), + ("a", 2, true, false), + ("b", 1, true, true), + ("b", 1, false, false), + ("b", 2, true, false), + ("a", 2, true, true), + ("a", 1, false, false), + ] { + let wrapper: Box = if tls { + Box::new(TlsClientHandler::new( + config.clone(), + None, + format!("{scope}.local").try_into().unwrap(), + Box::new(Carrier), + )) + } else { + Box::new(WebsocketTcpClientHandler::new( + Some(format!("/{scope}")), + None, + Default::default(), + Box::new(Carrier), + )) + }; + let socket = TcpStream::connect((std::net::Ipv4Addr::new(127, 0, 0, ip), port)) + .await + .unwrap(); + let location = ResolvedLocation::new(NetLocation::new( + Address::Ipv4(std::net::Ipv4Addr::LOCALHOST), + target, + )); + let carrier = wrapper + .setup_client_tcp_stream(Box::new(socket), location) + .await + .unwrap() + .client_stream; + let mut stream = client.handshake(carrier).await.unwrap(); + assert_eq!(stream.is_resumed(), resumed); + assert_eq!( + exchange(&mut stream, target).await.is_err(), + rejected, + "tls={tls}, scope={scope}, ip={ip}" + ); + } + listeners.stop().await; + destination.stop().await; + } + }) + .await + .unwrap(); +} diff --git a/src/vless/encryption/tests/transports.rs b/src/vless/encryption/tests/transports.rs index 37e31e03..c56542f8 100644 --- a/src/vless/encryption/tests/transports.rs +++ b/src/vless/encryption/tests/transports.rs @@ -17,7 +17,7 @@ use crate::{ const UUID: &str = "550e8400-e29b-41d4-a716-446655440000"; const LIMIT: Duration = Duration::from_secs(10); -struct Tasks(Vec>); +pub(super) struct Tasks(pub(super) Vec>); impl Drop for Tasks { fn drop(&mut self) { for task in &self.0 { @@ -26,7 +26,7 @@ impl Drop for Tasks { } } impl Tasks { - async fn stop(&mut self) { + pub(super) async fn stop(&mut self) { for task in self.0.drain(..) { task.abort(); let _ = task.await; @@ -34,7 +34,7 @@ impl Tasks { } } -async fn echo() -> (u16, Tasks) { +pub(super) async fn echo() -> (u16, Tasks) { let listener = TcpListener::bind("0.0.0.0:0").await.unwrap(); let port = listener.local_addr().unwrap().port(); let task = tokio::spawn(async move { @@ -55,7 +55,207 @@ async fn echo() -> (u16, Tasks) { (port, Tasks(vec![task])) } -fn request(port: u16, command: u8, flow: bool) -> Vec { +#[tokio::test] +async fn native_vision_rejects_h2mux_before_opening_a_stream() { + use crate::{ + client_proxy_selector::ClientProxySelector, + h2mux::{MUX_DESTINATION_HOST, MUX_DESTINATION_PORT}, + tcp::tcp_handler::TcpServerHandler, + vless::{encryption::ServerEncryption, vless_server_handler::VlessTcpServerHandler}, + }; + use shoes_test_support::vless::{VlessDestination, encode_vless_udp_request, parse_uuid}; + + let (encryption, decryption) = fixture_settings("random", true, &[32]); + let handler = VlessTcpServerHandler::new( + UUID, + true, + Arc::new(ClientProxySelector::new(Vec::new())), + Arc::new(NativeResolver::new()), + None, + Some(Arc::new( + ServerEncryption::new(&decryption, Default::default()).unwrap(), + )), + true, + ) + .unwrap(); + let (a, b) = tokio::io::duplex(8192); + let (result, _client) = timeout(LIMIT, async { + tokio::join!(handler.setup_server_stream(Box::new(b)), async { + let mut client = ClientEncryption::new(&encryption) + .unwrap() + .handshake(a) + .await + .unwrap(); + let mut request = encode_vless_udp_request( + parse_uuid(UUID).unwrap(), + &VlessDestination::Domain(MUX_DESTINATION_HOST.into(), MUX_DESTINATION_PORT), + ) + .unwrap(); + request[18] = 1; + request[17] = 18; + request.splice(18..18, b"\x0a\x10xtls-rprx-vision".iter().copied()); + client.write_all(&request).await.unwrap(); + client.flush().await.unwrap(); + client + }) + }) + .await + .unwrap(); + let error = result.err().expect("native Vision accepted H2MUX"); + assert_eq!(error.kind(), std::io::ErrorKind::InvalidData); + assert_eq!( + error.to_string(), + "Native VLESS Vision does not support H2MUX" + ); +} + +#[tokio::test] +async fn legacy_and_native_address_forms_and_empty_udp_responses() { + use crate::async_stream::AsyncStream; + use shoes_test_support::{ + test_servers::{start_tcp_stream_echo_server, start_udp_response_server}, + vless::{ + VlessDestination, VlessUdpClient, XUDP_OPTION_DATA, XUDP_STATUS_NEW, XudpClient, + XudpFrame, encode_vless_udp_request, parse_uuid, + }, + }; + + let udp_response = |payload: &[u8], _| { + if payload == b"empty reply" { + vec![] + } else { + payload.to_vec() + } + }; + let udp = start_udp_response_server("0.0.0.0", 0, udp_response) + .await + .unwrap(); + let (tcp_port, mut target) = echo().await; + // Separate listeners serve either localhost address without requiring dual-stack sockets. + let ipv6_targets = match TcpListener::bind("[::1]:0").await { + Ok(_probe) => Some(( + start_tcp_stream_echo_server("::1", tcp_port).await.unwrap(), + start_udp_response_server("::1", udp.local_addr().port(), udp_response) + .await + .unwrap(), + )), + Err(error) => { + eprintln!("IPv6 loopback unavailable; skipping IPv6 cases: {error}"); + None + } + }; + for (native, vision) in [(false, false), (true, false), (true, true)] { + let (encryption, decryption) = fixture_settings("random", true, &[32]); + let client = ClientEncryption::new(&encryption).unwrap(); + let port = TcpListener::bind("0.0.0.0:0") + .await + .unwrap() + .local_addr() + .unwrap() + .port(); + let config: ServerConfig = serde_json::from_value(json!({ + "address":format!("0.0.0.0:{port}"), + "protocol":{"type":"vless","user_id":UUID,"udp_enabled":true, + "decryption":if native {decryption.as_str()} else {"none"},"vision":vision} + })) + .unwrap(); + let mut listeners = Tasks( + crate::tcp::tcp_server::start_servers( + crate::config::Config::Server(config), + Arc::new(NativeResolver::new()), + ) + .await + .unwrap(), + ); + let connect = || async { + let socket = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + let stream: Box = if native { + Box::new(client.handshake(socket).await.unwrap()) + } else { + Box::new(socket) + }; + stream + }; + timeout(LIMIT, async { + let destinations = |port, ipv6: Option| { + let mut destinations = vec![ + VlessDestination::Ip(SocketAddr::from(([127, 0, 0, 1], port))), + VlessDestination::Domain("localhost".into(), port), + ]; + if let Some(address) = ipv6 { + destinations.push(VlessDestination::Ip(address)); + } + destinations + }; + for _ in 0..2 { + if !vision { + for destination in destinations( + tcp_port, + ipv6_targets.as_ref().map(|(tcp, _)| tcp.local_addr()), + ) { + let mut stream = connect().await; + let mut request = + encode_vless_udp_request(parse_uuid(UUID).unwrap(), &destination) + .unwrap(); + request[18] = 1; + stream.write_all(&request).await.unwrap(); + stream.write_all(b"address roundtrip").await.unwrap(); + stream.flush().await.unwrap(); + let mut reply = [0; 19]; + stream.read_exact(&mut reply).await.unwrap(); + assert_eq!(&reply[..2], &[0, 0]); + assert_eq!(&reply[2..], b"address roundtrip"); + } + } + for destination in destinations( + udp.local_addr().port(), + ipv6_targets.as_ref().map(|(_, udp)| udp.local_addr()), + ) { + let stream = connect().await; + let mut udp = VlessUdpClient::connect( + stream, + parse_uuid(UUID).unwrap(), + destination.clone(), + ) + .await + .unwrap(); + for (payload, expected) in [ + (b"empty reply".as_slice(), b"".as_slice()), + ( + b"nonempty after empty".as_slice(), + b"nonempty after empty".as_slice(), + ), + ] { + udp.send_packet(payload).await.unwrap(); + let reply = udp.recv_packet(LIMIT).await.unwrap(); + assert_eq!(reply, expected); + } + let mut xudp = XudpClient::connect(connect().await, parse_uuid(UUID).unwrap()) + .await + .unwrap(); + xudp.send_frame(&XudpFrame { + session_id: 17, + status: XUDP_STATUS_NEW, + options: XUDP_OPTION_DATA, + destination: Some(destination), + payload: Some(b"xudp address".to_vec()), + }) + .await + .unwrap(); + let reply = xudp.recv_frame(LIMIT).await.unwrap(); + assert_eq!(reply.session_id, 17); + assert_eq!(reply.payload.unwrap(), b"xudp address"); + } + } + }) + .await + .unwrap_or_else(|e| panic!("native={native}, vision={vision}: {e}")); + listeners.stop().await; + } + target.stop().await; +} + +pub(super) fn request(port: u16, command: u8, flow: bool) -> Vec { let mut request = vec![0]; request.extend(crate::uuid_util::parse_uuid(UUID).unwrap()); if flow { diff --git a/src/vless/vision_stream.rs b/src/vless/vision_stream.rs index cda99573..d22354a1 100644 --- a/src/vless/vision_stream.rs +++ b/src/vless/vision_stream.rs @@ -858,8 +858,11 @@ impl AsyncStream for VisionStream { #[cfg(test)] mod tests { mod decisions; + mod native_decisions; mod native_errors; #[cfg(target_os = "linux")] + mod native_splice; + #[cfg(target_os = "linux")] mod splice; use super::*; diff --git a/src/vless/vision_stream/tests/decisions.rs b/src/vless/vision_stream/tests/decisions.rs index 254c9fd7..f6511756 100644 --- a/src/vless/vision_stream/tests/decisions.rs +++ b/src/vless/vision_stream/tests/decisions.rs @@ -5,9 +5,9 @@ use tokio::time::{Duration, timeout}; type LegacyVision = VisionStream>; -const APPLICATION: &[u8] = b"\x17\x03\x03\x00\x03app"; +pub(super) const APPLICATION: &[u8] = b"\x17\x03\x03\x00\x03app"; const CHANGE_CIPHER_SPEC: &[u8] = b"\x14\x03\x03\x00\x01\x01"; -const TLS13_VERSION: &[u8] = b"\x00\x2b\x00\x02\x03\x04"; +pub(super) const TLS13_VERSION: &[u8] = b"\x00\x2b\x00\x02\x03\x04"; fn record(content_type: u8, body: &[u8]) -> Vec { let mut bytes = vec![content_type, 3, 3]; @@ -16,14 +16,14 @@ fn record(content_type: u8, body: &[u8]) -> Vec { bytes } -fn client_hello() -> Vec { +pub(super) fn client_hello() -> Vec { let (mut client, _) = new_connections(Backend::Tls13); let mut flight = Vec::new(); client.write_tls(&mut flight).unwrap(); flight } -fn server_hello(cipher: u16, session_id_len: u8, extensions: &[u8]) -> Vec { +pub(super) fn server_hello(cipher: u16, session_id_len: u8, extensions: &[u8]) -> Vec { let mut body = vec![3, 3]; body.extend_from_slice(&[42; 32]); body.push(session_id_len); @@ -69,7 +69,11 @@ async fn pair(backend: Backend) -> (LegacyVision, LegacyVision) { ) } -async fn transfer(sender: &mut LegacyVision, receiver: &mut LegacyVision, data: &[u8]) { +pub(super) async fn transfer( + sender: &mut VisionStream, + receiver: &mut VisionStream, + data: &[u8], +) { timeout(Duration::from_secs(2), async { sender.write_all(data).await.unwrap(); sender.flush().await.unwrap(); @@ -100,9 +104,9 @@ fn assert_terminal_write(stream: &VisionStream, direct: b } } -async fn assert_decision( - client: &mut LegacyVision, - server: &mut LegacyVision, +pub(super) async fn assert_decision( + client: &mut VisionStream, + server: &mut VisionStream, direct: bool, server_first: bool, ) { diff --git a/src/vless/vision_stream/tests/native_decisions.rs b/src/vless/vision_stream/tests/native_decisions.rs new file mode 100644 index 00000000..6f1fa264 --- /dev/null +++ b/src/vless/vision_stream/tests/native_decisions.rs @@ -0,0 +1,179 @@ +use super::decisions::{ + APPLICATION, TLS13_VERSION, assert_decision, client_hello, server_hello, transfer, +}; +use super::*; +use crate::vless::encryption::{ + ClientEncryption, EncryptionStream, ServerEncryption, tests::fixture_pair, +}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt, DuplexStream}, + time::{Duration, timeout}, +}; + +type NativeVision = VisionStream>; + +async fn pair( + client: &ClientEncryption, + server: &ServerEncryption, + resumed: bool, +) -> (NativeVision, NativeVision) { + let (a, b) = tokio::io::duplex(65536); + let (a, b) = timeout(Duration::from_secs(3), async { + tokio::join!( + async { + let mut stream = client.handshake(a).await?; + stream.flush().await?; + Ok::<_, io::Error>(stream) + }, + server.handshake(b) + ) + }) + .await + .unwrap(); + let a = a.unwrap(); + assert_eq!(a.is_resumed(), resumed); + ( + VisionStream::new_client(a, [7; 16]) + .unwrap() + .with_write_error_policy(VisionWriteErrorPolicy::AcknowledgeAccepted), + VisionStream::new_server(b.unwrap(), [7; 16], b"") + .unwrap() + .with_write_error_policy(VisionWriteErrorPolicy::AcknowledgeAccepted), + ) +} + +#[tokio::test] +async fn native_fresh_and_resumed_direct_decisions_preserve_both_directions() { + timeout(Duration::from_secs(60), async { + for mode in ["native", "xorpub", "random"] { + for cipher in [ + 0x1301, 0x1302, 0x1303, 0x1304, 0x1305, 0x1306, 0xc02f, 0x0a0a, + ] { + for extensions in [TLS13_VERSION, b"".as_slice()] { + for server_first in [false, true] { + let (client, server) = fixture_pair(mode, true, &[32]); + for resumed in [false, true] { + let (mut c, mut s) = pair(&client, &server, resumed).await; + transfer(&mut c, &mut s, &client_hello()).await; + transfer(&mut s, &mut c, &server_hello(cipher, 32, extensions)).await; + assert_decision( + &mut c, + &mut s, + (0x1301..=0x1304).contains(&cipher) && !extensions.is_empty(), + server_first, + ) + .await; + c.shutdown().await.unwrap(); + assert_eq!(s.read(&mut [0]).await.unwrap(), 0); + s.write_all(b"after half-close").await.unwrap(); + s.flush().await.unwrap(); + let mut reply = [0; 16]; + c.read_exact(&mut reply).await.unwrap(); + assert_eq!(&reply, b"after half-close"); + s.shutdown().await.unwrap(); + assert_eq!(c.read(&mut [0]).await.unwrap(), 0); + } + } + } + } + } + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn native_fragmented_records_preserve_direct_decisions_and_coalesced_tails() { + timeout(Duration::from_secs(60), async { + for mode in ["native", "xorpub", "random"] { + for cipher in [0x1301, 0x1305] { + let hello = server_hello(cipher, 32, TLS13_VERSION); + for split in 1..hello.len() { + let (client, server) = fixture_pair(mode, true, &[32]); + for resumed in [false, true] { + let (mut c, mut s) = pair(&client, &server, resumed).await; + transfer(&mut c, &mut s, &client_hello()).await; + s.write_all(&hello[..split]).await.unwrap(); + s.flush().await.unwrap(); + assert!(!s.filter.supports_xtls()); + s.write_all(&hello[split..]).await.unwrap(); + s.flush().await.unwrap(); + let mut received = vec![0; hello.len()]; + c.read_exact(&mut received).await.unwrap(); + assert_eq!(received, hello); + assert_decision(&mut c, &mut s, cipher == 0x1301, split % 2 == 0).await; + } + } + for split in 1..APPLICATION.len() { + let (client, server) = fixture_pair(mode, true, &[32]); + let (mut c, mut s) = pair(&client, &server, false).await; + transfer(&mut c, &mut s, &client_hello()).await; + transfer(&mut s, &mut c, &hello).await; + s.write_all(&APPLICATION[..split]).await.unwrap(); + s.flush().await.unwrap(); + let tail = [APPLICATION[split..].to_vec(), b"coalesced tail".to_vec()].concat(); + s.write_all(&tail).await.unwrap(); + s.flush().await.unwrap(); + let expected = [APPLICATION, b"coalesced tail"].concat(); + let mut received = vec![0; expected.len()]; + c.read_exact(&mut received).await.unwrap(); + assert_eq!(received, expected); + assert_eq!( + c.read_mode, + if cipher == 0x1301 { + VisionMode::Direct + } else { + VisionMode::Tls + } + ); + } + } + } + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn native_malformed_hello_and_non_tls_never_enable_direct() { + timeout(Duration::from_secs(60), async { + let mut short = server_hello(0x1301, 32, TLS13_VERSION); + short[6..9].copy_from_slice(&[0, 0, 2]); + let mut long = short.clone(); + long[6..9].copy_from_slice(&[0, 1, 0]); + let cases = [ + short, + long, + server_hello(0x1301, 33, TLS13_VERSION), + server_hello(0x1301, 0, b"\x00\x2b\x00\x01\x03"), + ]; + for mode in ["native", "xorpub", "random"] { + for hello in &cases { + let (client, server) = fixture_pair(mode, true, &[32]); + for resumed in [false, true] { + let (mut c, mut s) = pair(&client, &server, resumed).await; + transfer(&mut c, &mut s, &client_hello()).await; + transfer(&mut s, &mut c, hello).await; + assert_decision(&mut c, &mut s, false, false).await; + } + } + let (client, server) = fixture_pair(mode, true, &[32]); + for resumed in [false, true] { + let (mut c, mut s) = pair(&client, &server, resumed).await; + transfer(&mut c, &mut s, b"GET / HTTP/1.1\r\n\r\n").await; + transfer(&mut s, &mut c, b"HTTP/1.1 200 OK\r\n\r\n").await; + // Exhaust the shared read/write inspection budget, then commit END. + for _ in 0..10 { + transfer(&mut c, &mut s, &[b'x'; 256]).await; + transfer(&mut s, &mut c, &[b'y'; 256]).await; + } + assert_eq!(c.write_mode, VisionMode::Tls); + assert_eq!(s.read_mode, VisionMode::Tls); + assert!(!c.filter.supports_xtls()); + assert!(!s.filter.supports_xtls()); + } + } + }) + .await + .unwrap(); +} diff --git a/src/vless/vision_stream/tests/native_splice.rs b/src/vless/vision_stream/tests/native_splice.rs new file mode 100644 index 00000000..49b92ca4 --- /dev/null +++ b/src/vless/vision_stream/tests/native_splice.rs @@ -0,0 +1,125 @@ +use super::decisions::{APPLICATION, TLS13_VERSION, client_hello, server_hello}; +use super::*; +use crate::{copy_bidirectional::copy_bidirectional, vless::encryption::tests::fixture_pair}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + time::{Duration, timeout}, +}; + +async fn tcp_pair() -> (TcpStream, TcpStream) { + let listener = TcpListener::bind("0.0.0.0:0").await.unwrap(); + let (client, server) = tokio::join!( + TcpStream::connect(("127.0.0.1", listener.local_addr().unwrap().port())), + listener.accept() + ); + let client = client.unwrap(); + let server = server.unwrap().0; + client.set_nodelay(true).unwrap(); + server.set_nodelay(true).unwrap(); + (client, server) +} + +async fn transfer( + sender: &mut (impl AsyncStream + ?Sized), + receiver: &mut (impl AsyncStream + ?Sized), + bytes: &[u8], +) { + let mut received = vec![0; bytes.len()]; + let (sent, read) = tokio::join!( + async { + sender.write_all(bytes).await?; + sender.flush().await + }, + receiver.read_exact(&mut received) + ); + sent.unwrap(); + read.unwrap(); + assert_eq!(received, bytes); +} + +#[tokio::test] +async fn native_direct_splices_only_unmasked_tcp_and_preserves_half_close() { + timeout(Duration::from_secs(60), async { + for mode in ["native", "xorpub", "random"] { + for server_relay in [false, true] { + for hide_tcp in [false, true] { + let (client, server) = fixture_pair(mode, true, &[32]); + for resumed in [false, true] { + let (a, b) = tcp_pair().await; + let wrap = |socket| -> Box { + if hide_tcp { + Box::new(crate::prepend_stream::PrependStream::new(socket, None)) + } else { + Box::new(socket) + } + }; + let (a, b) = tokio::join!( + async { + let mut a = client.handshake(wrap(a)).await.unwrap(); + assert_eq!(a.is_resumed(), resumed); + a.flush().await.unwrap(); + a + }, + server.handshake(wrap(b)) + ); + let c = VisionStream::new_client(a, [7; 16]) + .unwrap() + .with_write_error_policy(VisionWriteErrorPolicy::AcknowledgeAccepted); + let s = VisionStream::new_server(b.unwrap(), [7; 16], b"") + .unwrap() + .with_write_error_policy(VisionWriteErrorPolicy::AcknowledgeAccepted); + let (mut peer, mut relay_io) = if server_relay { (c, s) } else { (s, c) }; + let (mut remote, mut target) = tcp_pair().await; + let relay = tokio::spawn(async move { + copy_bidirectional(&mut relay_io, &mut remote, false, false).await + }); + let hello = client_hello(); + let flight = [ + server_hello(0x1301, 32, TLS13_VERSION), + APPLICATION.to_vec(), + ] + .concat(); + if server_relay { + transfer(&mut peer, &mut target, &hello).await; + transfer(&mut target, &mut peer, &flight).await; + assert!(peer.plain_tcp().is_none()); + transfer(&mut peer, &mut target, APPLICATION).await; + } else { + transfer(&mut target, &mut peer, &hello).await; + transfer(&mut peer, &mut target, &flight).await; + assert!(peer.plain_tcp().is_none()); + transfer(&mut target, &mut peer, APPLICATION).await; + } + let expect_splice = mode != "random" && !hide_tcp; + assert_eq!(peer.plain_tcp().is_some(), expect_splice); + let payload: Vec<_> = + (0..1024 * 1024 + 37).map(|n| (n % 251) as u8).collect(); + for upload in [true, false] { + let before = crate::splice::TEST_SPLICED_BYTES.get(); + if upload { + transfer(&mut peer, &mut target, &payload).await; + } else { + transfer(&mut target, &mut peer, &payload).await; + } + let copied = crate::splice::TEST_SPLICED_BYTES.get() - before; + assert_eq!( + copied > 0, + expect_splice, + "{mode}, server={server_relay}, hidden={hide_tcp}, resumed={resumed}, upload={upload}" + ); + } + peer.shutdown().await.unwrap(); + assert_eq!(target.read(&mut [0]).await.unwrap(), 0); + transfer(&mut target, &mut peer, b"response after EOF").await; + target.shutdown().await.unwrap(); + assert_eq!(peer.read(&mut [0]).await.unwrap(), 0); + relay.await.unwrap().unwrap(); + } + } + } + } + }) + .await + .unwrap(); +} diff --git a/src/xhttp/carrier.rs b/src/xhttp/carrier.rs index c3ee0904..1d581097 100644 --- a/src/xhttp/carrier.rs +++ b/src/xhttp/carrier.rs @@ -350,6 +350,7 @@ impl Carrier { Sender::H2(sender) => { let (parts, body) = request.into_parts(); let empty = hyper::body::Body::is_end_stream(&body); + let streaming = matches!(&body, Body::Stream { .. }); let mut sender = tokio::time::timeout(timeouts.headers, sender.clone().ready()) .await? .map_err(io::Error::other)?; @@ -365,10 +366,16 @@ impl Carrier { let _carrier = carrier; let result = tokio::select! { biased; - _ = request_cancel.cancelled() => Err(io::ErrorKind::ConnectionAborted.into()), + _ = request_cancel.cancelled() => Err(H2SendError::Local(io::ErrorKind::ConnectionAborted.into())), result = send_h2_tracked(body, &mut send, upload) => result, }; if let Err(error) = result { + // RFC 9113 section 8.1 allows an early response followed by NO_ERROR. + // Drain the response; its END_STREAM, not this reset, establishes EOF. + if streaming && error.peer_sent_no_error(&mut send) { + return; + } + let error = error.into_io(); log::debug!("XHTTP H2 upload failed: {error}"); send.send_reset(h2::Reason::CANCEL); cancel.cancel(); @@ -551,41 +558,70 @@ pub async fn receive_h2(mut body: h2::RecvStream, mut writer: Writer) { } pub async fn send_h2(body: Body, mut send: h2::SendStream) -> io::Result<()> { - send_h2_tracked(body, &mut send, None).await + send_h2_tracked(body, &mut send, None) + .await + .map_err(H2SendError::into_io) +} + +#[derive(Debug)] +enum H2SendError { + Local(io::Error), + Stream(io::Error), +} + +impl H2SendError { + fn into_io(self) -> io::Error { + match self { + Self::Local(error) | Self::Stream(error) => error, + } + } + + fn peer_sent_no_error(&self, send: &mut h2::SendStream) -> bool { + // poll_capacity can return None and send_data can return a local stream-state error. + // Consult the stream's reason without hiding errors from the body producer. + matches!(self, Self::Stream(_)) + && matches!( + send.poll_reset(&mut std::task::Context::from_waker(std::task::Waker::noop())), + std::task::Poll::Ready(Ok(h2::Reason::NO_ERROR)) + ) + } +} + +impl From for H2SendError { + fn from(error: h2::Error) -> Self { + Self::Stream(io::Error::other(error)) + } } async fn send_h2_tracked( mut body: Body, send: &mut h2::SendStream, mut upload: Option, -) -> io::Result<()> { +) -> Result<(), H2SendError> { 'frames: loop { let frame = tokio::select! { frame = body.frame() => frame, - reset = futures::future::poll_fn(|cx| send.poll_reset(cx)) => return Err(io::Error::other(format!("XHTTP H2 stream reset: {reset:?}"))), + reset = futures::future::poll_fn(|cx| send.poll_reset(cx)) => return Err(H2SendError::Stream(io::Error::other(format!("XHTTP H2 stream reset: {reset:?}")))), }; let Some(frame) = frame else { - send.send_data(Bytes::new(), true) - .map_err(io::Error::other)?; + send.send_data(Bytes::new(), true)?; break; }; - let Ok(mut data) = frame?.into_data() else { + let Ok(mut data) = frame.map_err(H2SendError::Local)?.into_data() else { continue; }; while !data.is_empty() { send.reserve_capacity(data.len()); let capacity = futures::future::poll_fn(|cx| send.poll_capacity(cx)) .await - .ok_or_else(|| io::Error::from(io::ErrorKind::BrokenPipe))? - .map_err(io::Error::other)?; + .ok_or_else(|| H2SendError::Stream(io::ErrorKind::BrokenPipe.into()))??; let len = capacity.min(data.len()); if len == 0 { continue; } // Known-length peers may respond and reset immediately after the final bytes. let end = len == data.len() && hyper::body::Body::is_end_stream(&body); - send.send_data(data.split_to(len), end) - .map_err(io::Error::other)?; + send.send_data(data.split_to(len), end)?; if let Some(upload) = &upload { upload.activity.touch(); } @@ -603,6 +639,344 @@ async fn send_h2_tracked( #[cfg(test)] mod tests { use super::*; + use crate::resources::Budget; + use tokio::io::AsyncReadExt; + + async fn h2_peer() -> ( + Arc, + tokio::sync::mpsc::UnboundedReceiver<( + Request, + h2::server::SendResponse, + )>, + Tasks, + ) { + let (client, peer) = tokio::io::duplex(4096); + let tasks = Tasks::new(); + let (requests, received) = tokio::sync::mpsc::unbounded_channel(); + tasks.spawn(async move { + let mut connection = h2::server::Builder::new() + .initial_window_size(1024) + .handshake(peer) + .await + .unwrap(); + while let Some(request) = connection.accept().await { + if requests.send(request.unwrap()).is_err() { + break; + } + } + }); + let carrier = Carrier::tcp(Box::new(client), XhttpVersion::H2, 8192) + .await + .unwrap(); + (carrier, received, tasks) + } + + #[tokio::test] + async fn h2_early_response_drains_after_no_error_reset() { + tokio::time::timeout(Duration::from_secs(5), async { + for blocked_upload in [false, true] { + for response_len in [0, 48 * 1024] { + let (carrier, mut requests, _server) = h2_peer().await; + let tasks = Tasks::new(); + let budget = Arc::new(Budget::new(Some(16 * 1024))); + let memory = Memory::new(budget.clone(), 16 * 1024); + let (mut upload, body) = pipe(memory.clone()); + let data = vec![7; if blocked_upload { 2048 } else { 1024 }]; + upload.write_all(&data).await.unwrap(); + let expected: Vec<_> = (0..response_len).map(|i| (i % 251) as u8).collect(); + let request = Request::post("http://localhost/stream-one") + .body(Body::stream(body)) + .unwrap(); + let (response, ()) = tokio::join!( + carrier.request(request, memory.clone(), &tasks, Duration::from_secs(1)), + async { + let (request, mut respond) = requests.recv().await.unwrap(); + let mut incoming = request.into_body(); + assert_eq!(incoming.data().await.unwrap().unwrap(), &data[..1024]); + // Leave the upload open, including its exhausted window in the blocked case. + let mut send = respond + .send_response(Response::new(()), expected.is_empty()) + .unwrap(); + if !expected.is_empty() { + send.send_data(Bytes::copy_from_slice(&expected), true) + .unwrap(); + } + // h2 queues NO_ERROR after the complete response when these handles drop. + } + ); + let mut response = response.unwrap().into_body(); + upload.closed().await; + assert!(upload.write_all(b"after reset").await.is_err()); + let mut actual = Vec::new(); + response.read_to_end(&mut actual).await.unwrap(); + assert_eq!(actual, expected); + tasks.wait().await; + assert!(!tasks.cancel.is_cancelled()); + drop(response); + drop(upload); + assert_eq!(budget.snapshot().active, 0); + + // The reset belongs to one stream, not the pooled connection. + let probe_tasks = Tasks::new(); + let (probe, ()) = tokio::join!( + carrier.request( + Request::get("http://localhost/probe") + .body(Body::Empty) + .unwrap(), + memory, + &probe_tasks, + Duration::from_secs(1), + ), + async { + let (_, mut respond) = requests.recv().await.unwrap(); + let mut send = respond.send_response(Response::new(()), false).unwrap(); + send.send_data(Bytes::from_static(b"reused"), true).unwrap(); + } + ); + let mut actual = Vec::new(); + probe + .unwrap() + .into_body() + .read_to_end(&mut actual) + .await + .unwrap(); + assert_eq!(actual, b"reused"); + probe_tasks.wait().await; + assert_eq!(budget.snapshot().active, 0); + } + } + }) + .await + .unwrap(); + } + + #[tokio::test] + async fn h2_upload_failures_and_truncated_responses_are_not_eof() { + enum Failure { + Reset(h2::Reason), + Body, + Cancel, + Disconnect, + } + tokio::time::timeout(Duration::from_secs(5), async { + for failure in [ + Failure::Reset(h2::Reason::NO_ERROR), + Failure::Reset(h2::Reason::CANCEL), + Failure::Reset(h2::Reason::INTERNAL_ERROR), + Failure::Body, + Failure::Cancel, + Failure::Disconnect, + ] { + let (carrier, mut requests, server) = h2_peer().await; + let tasks = Tasks::new(); + let memory = Memory::new(Arc::new(Budget::new(Some(16 * 1024))), 16 * 1024); + let (mut upload, body) = pipe(memory.clone()); + let request = Request::post("http://localhost/stream-one") + .body(Body::stream(body)) + .unwrap(); + let (response, mut send) = tokio::join!( + carrier.request(request, memory, &tasks, Duration::from_secs(1)), + async { + let (_request, mut respond) = requests.recv().await.unwrap(); + let mut send = respond.send_response(Response::new(()), false).unwrap(); + send.send_data(Bytes::from_static(b"prefix"), false) + .unwrap(); + send + } + ); + let mut response = response.unwrap().into_body(); + let mut prefix = [0; 6]; + response.read_exact(&mut prefix).await.unwrap(); + assert_eq!(&prefix, b"prefix"); + match failure { + Failure::Reset(reason) => send.send_reset(reason), + Failure::Body => upload.fail(io::ErrorKind::UnexpectedEof.into()).await, + Failure::Cancel => tasks.cancel.cancel(), + Failure::Disconnect => server.cancel.cancel(), + } + upload.closed().await; + assert!(response.read(&mut [0; 1]).await.is_err()); + tasks.wait().await; + let quiet = matches!(failure, Failure::Reset(h2::Reason::NO_ERROR)); + assert_eq!(tasks.cancel.is_cancelled(), !quiet); + } + }) + .await + .unwrap(); + } + + #[tokio::test] + async fn h2_no_error_goaway_does_not_complete_a_truncated_response() { + use bytes::{BufMut, BytesMut}; + + fn frame(kind: u8, flags: u8, stream: u32, payload: &[u8]) -> Bytes { + let mut frame = BytesMut::new(); + frame.put_uint(payload.len() as u64, 3); + frame.put_u8(kind); + frame.put_u8(flags); + frame.put_u32(stream); + frame.extend_from_slice(payload); + frame.freeze() + } + + tokio::time::timeout(Duration::from_secs(5), async { + let (client, mut peer) = tokio::io::duplex(4096); + let server = Tasks::new(); + let shutdown = CancellationToken::new(); + let stop = shutdown.clone(); + let release_peer = CancellationToken::new(); + let release = release_peer.clone(); + server.spawn(async move { + let mut preface = [0; 24]; + peer.read_exact(&mut preface).await.unwrap(); + assert_eq!(&preface, b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"); + peer.write_all(&frame(4, 0, 0, &[])).await.unwrap(); + loop { + let mut header = [0; 9]; + peer.read_exact(&mut header).await.unwrap(); + let len = u32::from_be_bytes([0, header[0], header[1], header[2]]) as usize; + let mut payload = vec![0; len]; + peer.read_exact(&mut payload).await.unwrap(); + match (header[3], header[4] & 1) { + // Acknowledge SETTINGS and the carrier's readiness PING. + (4, 0) => peer.write_all(&frame(4, 1, 0, &[])).await.unwrap(), + (6, 0) => peer.write_all(&frame(6, 1, 0, &payload)).await.unwrap(), + (1, _) => { + assert_eq!(&header[5..], &1_u32.to_be_bytes()); + break; + } + _ => {} + } + } + // HEADERS (:status 200) and DATA without END_STREAM. + peer.write_all(&frame(1, 4, 1, &[0x88])).await.unwrap(); + peer.write_all(&frame(0, 0, 1, b"prefix")).await.unwrap(); + stop.cancelled().await; + // Exclude stream 1 from GOAWAY so it receives NO_ERROR, not transport EOF. + peer.write_all(&frame(7, 0, 0, &[0; 8])).await.unwrap(); + release.cancelled().await; + drop(peer); + }); + let carrier = Carrier::tcp(Box::new(client), XhttpVersion::H2, 8192) + .await + .unwrap(); + let tasks = Tasks::new(); + let memory = Memory::new(Arc::new(Budget::new(Some(16 * 1024))), 16 * 1024); + let (upload, body) = pipe(memory.clone()); + let mut response = carrier + .request( + Request::post("http://localhost/stream-one") + .body(Body::stream(body)) + .unwrap(), + memory, + &tasks, + Duration::from_secs(1), + ) + .await + .unwrap() + .into_body(); + let mut prefix = [0; 6]; + response.read_exact(&mut prefix).await.unwrap(); + assert_eq!(&prefix, b"prefix"); + shutdown.cancel(); + let error = response.read(&mut [0; 1]).await.unwrap_err(); + let error = error + .get_ref() + .unwrap() + .downcast_ref::() + .unwrap(); + assert!(error.is_go_away()); + assert_eq!(error.reason(), Some(h2::Reason::NO_ERROR)); + upload.closed().await; + tasks.wait().await; + assert!(!tasks.cancel.is_cancelled()); + release_peer.cancel(); + server.wait().await; + }) + .await + .unwrap(); + } + + #[tokio::test] + async fn h2_early_response_does_not_acknowledge_incomplete_packet() { + tokio::time::timeout(Duration::from_secs(5), async { + let (carrier, mut requests, _server) = h2_peer().await; + let tasks = Tasks::new(); + let budget = Arc::new(Budget::new(Some(16 * 1024))); + let memory = Memory::new(budget.clone(), 16 * 1024); + let (signal, written) = UploadSignal::channel(); + let request = Request::post("http://localhost/packet/0") + .extension(signal) + .body(Body::Chunks(vec![memory.copy(&[7; 2048]).unwrap()].into())) + .unwrap(); + let reservation = carrier.reserve().await.unwrap(); + let (response, ()) = tokio::join!( + carrier.request_reserved( + request, + memory, + &tasks, + Timeouts::packet(Duration::from_secs(1), Duration::from_secs(1)), + reservation, + ), + async { + let (request, mut respond) = requests.recv().await.unwrap(); + let mut incoming = request.into_body(); + assert_eq!(incoming.data().await.unwrap().unwrap().len(), 1024); + respond.send_response(Response::new(()), true).unwrap(); + } + ); + assert!(response.is_err()); + assert!(written.await.is_err()); + tasks.wait().await; + assert!(tasks.cancel.is_cancelled()); + assert_eq!(budget.snapshot().active, 0); + }) + .await + .unwrap(); + } + + #[tokio::test] + async fn h2_no_error_does_not_hide_body_errors_or_finish_server_sends() { + tokio::time::timeout(Duration::from_secs(5), async { + let (carrier, mut requests, _server) = h2_peer().await; + let Sender::H2(sender) = &carrier.sender else { + unreachable!() + }; + let mut sender = sender.clone().ready().await.unwrap(); + let (response, mut send) = sender + .send_request( + Request::post("http://localhost/stream-one") + .body(()) + .unwrap(), + false, + ) + .unwrap(); + let (request, mut respond) = requests.recv().await.unwrap(); + respond.send_response(Response::new(()), true).unwrap(); + drop((request, respond)); + let reason = futures::future::poll_fn(|cx| send.poll_reset(cx)) + .await + .unwrap(); + assert_eq!(reason, h2::Reason::NO_ERROR); + + assert!( + futures::future::poll_fn(|cx| send.poll_capacity(cx)) + .await + .is_none() + ); + let error = H2SendError::Stream(io::ErrorKind::BrokenPipe.into()); + assert!(error.peer_sent_no_error(&mut send)); + // A reset can arrive between the body EOF poll and sending the final DATA frame. + let error = H2SendError::from(send.send_data(Bytes::new(), true).unwrap_err()); + assert!(error.peer_sent_no_error(&mut send)); + let error = H2SendError::Local(io::ErrorKind::UnexpectedEof.into()); + assert!(!error.peer_sent_no_error(&mut send)); + assert!(send_h2(Body::Empty, send).await.is_err()); + assert!(response.await.unwrap().into_body().data().await.is_none()); + }) + .await + .unwrap(); + } #[tokio::test(start_paused = true)] async fn packet_response_wait_requires_upload_completion_and_observes_cancellation() { diff --git a/src/xhttp/lifecycle_tests.rs b/src/xhttp/lifecycle_tests.rs index 9c0ac8c3..d2c895c4 100644 --- a/src/xhttp/lifecycle_tests.rs +++ b/src/xhttp/lifecycle_tests.rs @@ -211,6 +211,73 @@ async fn delayed_headers_half_close_and_pool_isolation() { } } +#[tokio::test] +async fn h2_streaming_early_responses_preserve_download_tails() { + for mode in [XhttpMode::StreamOne, XhttpMode::StreamUp] { + tokio::time::timeout(Duration::from_secs(5), async { + let listener = tokio::net::TcpListener::bind("0.0.0.0:0").await.unwrap(); + let address = ([127, 0, 0, 1], listener.local_addr().unwrap().port()).into(); + let tasks = Tasks::new(); + let expected: Vec<_> = (0..96 * 1024).map(|i| (i % 251) as u8).collect(); + let response_bytes = expected.clone(); + tasks.spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut connection = h2::server::Builder::new() + .initial_window_size(1024) + .handshake(socket) + .await + .unwrap(); + let requests = Tasks::new(); + let uploaded = tokio_util::sync::CancellationToken::new(); + while let Some(Ok((request, mut respond))) = connection.accept().await { + let uploaded = uploaded.clone(); + let response_bytes = response_bytes.clone(); + requests.spawn(async move { + let post = request.method() == http::Method::POST; + let mut incoming = request.into_body(); + if post { + assert_eq!(incoming.data().await.unwrap().unwrap(), &[7; 1024][..]); + uploaded.cancel(); + } else { + uploaded.cancelled().await; + } + let upload_only = post && mode == XhttpMode::StreamUp; + let mut send = respond + .send_response(Response::new(()), upload_only) + .unwrap(); + if !upload_only { + send.send_data(Bytes::from(response_bytes), true).unwrap(); + } + // Dropping the unread POST queues NO_ERROR after its response. + }); + } + }); + let config = serde_json::from_value(serde_json::json!({ + "limits":{"max_buffered_bytes":65536,"max_session_buffered_bytes":65536}, + "protocol":{"type":"socks"} + })) + .unwrap(); + let client = configured_client(config, XhttpVersion::H2, mode); + let dialer = Arc::new(Dialer { + address, + key: RouteKey(vec![1]), + opens: AtomicUsize::new(0), + }); + let mut stream = client.open(dialer.clone()).await.unwrap(); + stream.write_all(&[7; 2048]).await.unwrap(); + stream.write.closed().await; + let mut actual = Vec::new(); + stream.read_to_end(&mut actual).await.unwrap(); + assert_eq!(actual, expected); + stream._tasks.wait().await; + assert!(stream.write_all(b"after response").await.is_err()); + assert_eq!(dialer.opens.load(Ordering::Relaxed), 1); + }) + .await + .unwrap(); + } +} + #[tokio::test] async fn idle_routes_are_evicted_and_failed_dials_do_not_fill_the_pool() { let (address, _server) = echo_peer(XhttpVersion::H2, false).await; diff --git a/src/xhttp/server.rs b/src/xhttp/server.rs index dbd93f61..54413e6b 100644 --- a/src/xhttp/server.rs +++ b/src/xhttp/server.rs @@ -375,15 +375,23 @@ mod tests { state.close(suffix); } let mut data = Vec::new(); - let _ = + let result = tokio::time::timeout(Duration::from_secs(1), response.read_to_end(&mut data)) .await .unwrap(); assert!(data.is_empty()); - // H3's idle sender observes STOP_SENDING on its ten-second probe. - tokio::time::timeout(Duration::from_secs(12), upload.cancel.cancelled()) - .await - .unwrap(); + if version == XhttpVersion::H2 { + result.unwrap(); + tokio::time::timeout(Duration::from_secs(1), upload.wait()) + .await + .unwrap(); + assert!(!upload.cancel.is_cancelled()); + } else { + // H3's idle sender observes STOP_SENDING on its ten-second probe. + tokio::time::timeout(Duration::from_secs(12), upload.cancel.cancelled()) + .await + .unwrap(); + } let next = Tasks::new(); let response = tokio::time::timeout( Duration::from_secs(2), diff --git a/tests/XHTTP.md b/tests/XHTTP.md index e2caa440..02cb5b7d 100644 --- a/tests/XHTTP.md +++ b/tests/XHTTP.md @@ -3,7 +3,7 @@ The `xhttp_interop` integration target runs real Shoes processes against Shoes, Xray-core, and Mihomo Alpha. CI builds and caches these peers from pinned source: -- Xray-core: `701af60772cda123492f86e383e1fdba066614a9` (26.10.10). +- Xray-core: `92ccc939aaf4f8965f369cf0b142e005aeaa11d5` (26.10.17). - Mihomo Alpha: `94b13aed49b8dc41973e4bb3f721c5fa2c5c5c1f`. Use Go 1.27.1, `GOMAXPROCS=1`, and `go build -p 1`. Build Xray's `./main` package. @@ -113,10 +113,10 @@ This extended sweep is opt-in; normal CI runs the focused recovery/containment c ### Local Stall Qualification -On 2026-10-16, the pinned Xray client completed all six 16 MiB duplex cases against -Shoes with 32 KiB POSTs, 30 ms pacing, no added RTT, and destination pauses of -10/30/60 seconds. These correctness runs used a debug Shoes binary; their rates -are not production throughput measurements. +On 2026-10-16, the then-pinned Xray 26.10.10 (`701af607`) client completed all six +16 MiB duplex cases against Shoes with 32 KiB POSTs, 30 ms pacing, no added RTT, +and destination pauses of 10/30/60 seconds. These correctness runs used a debug +Shoes binary; their rates are not production throughput measurements. | Server request cap | Total duration, 10/30/60 s pause | Peak service handlers | Peak application bytes | Combined peer RSS range | | --- | --- | --- | --- | --- | @@ -137,8 +137,9 @@ for Shoes-to-Xray. Each session resumed with exact upload and download bytes. ### Local Throughput Qualification -The release Shoes binary passed all 24 combinations of 0/50/150/300 ms added RTT, -32 KiB/1,000,000-byte POST limits, and the three peer directions. All used 30 ms +Using the then-pinned Xray 26.10.10 (`701af607`), the release Shoes binary passed +all 24 combinations of 0/50/150/300 ms added RTT, 32 KiB/1,000,000-byte POST limits, +and the three peer directions. All used 30 ms pacing and exact 16 MiB transfers each way; Shoes used its default client pipeline depth of 30. Representative aggregate duplex rates for 1,000,000-byte POSTs: diff --git a/tests/cases/optional_peer/tls_hybrid.rs b/tests/cases/optional_peer/tls_hybrid.rs index 902c82c5..486573b0 100644 --- a/tests/cases/optional_peer/tls_hybrid.rs +++ b/tests/cases/optional_peer/tls_hybrid.rs @@ -13,7 +13,7 @@ use tokio::{ fn start_xray(config: serde_json::Value) -> io::Result<(ProcessGuard, tempfile::NamedTempFile)> { let binary = std::env::var_os("SHOES_TEST_XRAY_BIN").unwrap_or_else(|| "xray".into()); - let mut file = tempfile::NamedTempFile::new()?; + let mut file = tempfile::Builder::new().suffix(".json").tempfile()?; file.write_all(config.to_string().as_bytes())?; file.flush()?; let child = Command::new(binary) diff --git a/tests/cases/optional_peer/vless_encryption.rs b/tests/cases/optional_peer/vless_encryption.rs index b7dde434..2fbfdb70 100644 --- a/tests/cases/optional_peer/vless_encryption.rs +++ b/tests/cases/optional_peer/vless_encryption.rs @@ -23,14 +23,21 @@ const LIMIT: Duration = Duration::from_secs(10); #[path = "vless_encryption/benchmarks.rs"] mod benchmarks; +#[path = "vless_encryption/chained.rs"] +mod chained; +#[path = "vless_encryption/composition.rs"] +mod composition; #[path = "vless_encryption/legacy.rs"] mod legacy; +#[path = "vless_encryption/lifecycle.rs"] +mod lifecycle; #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum Kind { Shoes, Xray, Mihomo, + SingBox, } impl Kind { @@ -45,6 +52,12 @@ impl Kind { Self::Mihomo => std::env::var_os("MIHOMO_BINARY") .expect("MIHOMO_BINARY must name the pinned Mihomo binary; no silent skips") .into(), + Self::SingBox => { + std::env::var_os("SHOES_TEST_SING_BOX_BIN").expect( + "SHOES_TEST_SING_BOX_BIN must name the pinned sing-box binary; no silent skips", + ); + shoes_test_support::process::find_singbox_binary().unwrap() + } } } } @@ -70,6 +83,20 @@ impl Peer { fn logs(&self) -> String { std::fs::read_to_string(self.directory.path().join("peer.log")).unwrap_or_default() } + + fn assert_handshakes(&self, expected: &[&str]) { + let logs = self.logs(); + let actual: Vec<_> = logs + .lines() + .filter_map(|line| { + let (_, handshake) = line + .split_once("VLESS encryption client: ") + .or_else(|| line.split_once("VLESS encryption server: "))?; + handshake.strip_suffix(" handshake") + }) + .collect(); + assert_eq!(actual, expected, "{logs}"); + } } fn pair_logs(client: &Peer, server: &Peer) -> String { @@ -88,7 +115,17 @@ async fn start(kind: Kind, port: u16, config: Value) -> Peer { start_binary(kind, port, config, kind.binary()).await } -async fn start_binary(kind: Kind, port: u16, mut config: Value, binary: PathBuf) -> Peer { +async fn start_binary(kind: Kind, port: u16, config: Value, binary: PathBuf) -> Peer { + start_options(kind, port, config, binary, None).await +} + +async fn start_options( + kind: Kind, + port: u16, + mut config: Value, + binary: PathBuf, + go_debug: Option<&str>, +) -> Peer { // Mihomo opens listeners before its tunnel becomes Running. A separate // direct route probes readiness without warming any native ticket cache. let mut readiness = if kind == Kind::Mihomo { @@ -118,15 +155,19 @@ async fn start_binary(kind: Kind, port: u16, mut config: Value, binary: PathBuf) let mut command = Command::new(binary); match kind { Kind::Shoes => { - command.args(["--threads", "1", "--no-reload"]).arg(&path); + command.args(["--threads", "1", "--no-reload"]); + command.arg(&path); } - Kind::Xray => { + Kind::Xray | Kind::SingBox => { command.args(["run", "-c"]).arg(&path); } Kind::Mihomo => { command.arg("-d").arg(directory.path()).arg("-f").arg(&path); } } + if let Some(value) = go_debug { + command.env("GODEBUG", value); + } let child = command .env("GOMAXPROCS", "1") .env("RUST_LOG", "shoes=debug") @@ -175,12 +216,33 @@ async fn start_binary(kind: Kind, port: u16, mut config: Value, binary: PathBuf) } async fn free_port() -> u16 { - TcpListener::bind("0.0.0.0:0") - .await - .unwrap() - .local_addr() - .unwrap() - .port() + static ALLOCATED: Mutex> = Mutex::new(Vec::new()); + for _ in 0..100 { + let tcp = TcpListener::bind("0.0.0.0:0").await.unwrap(); + let port = tcp.local_addr().unwrap().port(); + // SOCKS and QUIC peers also bind UDP on this port. + match tokio::net::UdpSocket::bind(("0.0.0.0", port)).await { + Ok(_udp) => { + let mut allocated = ALLOCATED.lock().unwrap(); + if allocated.contains(&port) { + continue; + } + allocated.push(port); + return port; + } + Err(error) if error.kind() == io::ErrorKind::AddrInUse => continue, + Err(error) => panic!("reserve UDP port: {error}"), + } + } + panic!("could not find a free TCP/UDP port"); +} + +#[tokio::test] +async fn allocated_peer_ports_are_not_reused_before_listeners_start() { + let mut ports = std::collections::HashSet::new(); + for _ in 0..512 { + assert!(ports.insert(free_port().await)); + } } fn credentials(mode: &str, auth: &[&str]) -> (String, String, usize) { @@ -231,6 +293,11 @@ fn server_config(kind: Kind, port: u16, decryption: &str, vision: bool) -> Value "listeners": [{"name": "encrypted", "type": "vless", "listen": "0.0.0.0", "port": port, "decryption": decryption, "users": [{"username": "test", "uuid": USER, "flow": flow}]}], }), + Kind::SingBox => { + assert_eq!(decryption, "none"); + json!({"log":{"level":"debug"},"inbounds":[{"type":"vless","listen":"0.0.0.0","listen_port":port, + "users":[{"uuid":USER,"flow":flow}]}],"outbounds":[{"type":"direct"}]}) + } } } @@ -255,6 +322,11 @@ fn client_config(kind: Kind, port: u16, server_port: u16, encryption: &str, visi "uuid": USER, "encryption": encryption, "flow": flow, "udp": true, "packet-encoding": "packetaddr"}], "mode": "rule", "rules": ["MATCH,encrypted"], }), + Kind::SingBox => { + assert_eq!(encryption, "none"); + json!({"log":{"level":"debug"},"inbounds":[{"type":"socks","listen":"0.0.0.0","listen_port":port}], + "outbounds":[{"type":"vless","server":"127.0.0.1","server_port":server_port,"uuid":USER,"flow":flow}]}) + } } } @@ -596,7 +668,13 @@ enum Transport { Raw, Tls, Websocket, + Wss, Xhttp, + XhttpMode { + version: &'static str, + mode: &'static str, + }, + Quic, Reality, } @@ -616,6 +694,21 @@ fn xray_endpoint_mut(config: &mut Value, server: bool) -> &mut Value { } } +fn reality_keys() -> (String, String) { + use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; + + let private = [42; 32]; + let public = + aws_lc_rs::agreement::PrivateKey::from_private_key(&aws_lc_rs::agreement::X25519, &private) + .unwrap() + .compute_public_key() + .unwrap(); + ( + URL_SAFE_NO_PAD.encode(private), + URL_SAFE_NO_PAD.encode(public.as_ref()), + ) +} + fn wrap( kind: Kind, mut config: Value, @@ -626,6 +719,64 @@ fn wrap( cover_port: u16, ) -> Value { match (kind, transport) { + (Kind::SingBox, _) => { + return legacy::wrap_singbox(config, server, transport, cert, key, cover_port); + } + (_, Transport::Wss) => { + config = wrap( + kind, + config, + server, + Transport::Websocket, + cert, + key, + cover_port, + ); + // TLS wraps the WebSocket stream, not the inner VLESS stream. + let ws = if kind == Kind::Xray { + Some(xray_endpoint_mut(&mut config, server)["streamSettings"].clone()) + } else { + None + }; + config = wrap(kind, config, server, Transport::Tls, cert, key, cover_port); + if let Some(ws) = ws { + let stream = &mut xray_endpoint_mut(&mut config, server)["streamSettings"]; + stream["network"] = ws["network"].clone(); + stream["wsSettings"] = ws["wsSettings"].clone(); + } + } + (_, Transport::XhttpMode { version, mode }) => { + return composition::wrap_xhttp( + kind, + config, + server, + (version, mode), + cert, + key, + cover_port, + ); + } + (Kind::Shoes, Transport::Quic) => { + let endpoint = if server { + &mut config[0] + } else { + &mut config[0]["rules"][0]["client_chain"] + }; + endpoint["transport"] = json!("quic"); + endpoint["quic_settings"] = if server { + json!({"cert": cert, "key": key}) + } else { + json!({"verify": false, "sni_hostname": "localhost"}) + }; + if server { + let address = config[0]["address"].clone(); + config + .as_array_mut() + .unwrap() + .push(json!({"address":address,"protocol":{"type":"socks"}})); + } + } + (_, Transport::Quic) => unreachable!("raw QUIC is a Shoes transport"), (_, Transport::Raw) => {} (Kind::Shoes, Transport::Tls) => { let field = shoes_protocol_mut(&mut config, server); @@ -694,17 +845,7 @@ fn wrap( field["streamSettings"] = json!({"network": "xhttp", "security": "none", "xhttpSettings": {"path": "/native", "host": "localhost", "mode": "packet-up"}}); } (Kind::Shoes | Kind::Xray, Transport::Reality) => { - use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; - let private = [42; 32]; - let public = aws_lc_rs::agreement::PrivateKey::from_private_key( - &aws_lc_rs::agreement::X25519, - &private, - ) - .unwrap() - .compute_public_key() - .unwrap(); - let public = URL_SAFE_NO_PAD.encode(public.as_ref()); - let private = URL_SAFE_NO_PAD.encode(private); + let (private, public) = reality_keys(); if kind == Kind::Shoes { let field = shoes_protocol_mut(&mut config, server); let inner = field.take(); @@ -724,8 +865,31 @@ fn wrap( json!({"network": "raw", "security": "reality", "realitySettings": reality}); } } - (Kind::Mihomo, Transport::Xhttp | Transport::Reality) => { - unreachable!("targeted Xray-only transport cases") + (Kind::Mihomo, Transport::Xhttp) => { + return composition::wrap_xhttp( + kind, + config, + server, + ("plain", "packet-up"), + cert, + key, + cover_port, + ); + } + (Kind::Mihomo, Transport::Reality) => { + let (private, public) = reality_keys(); + if server { + config["listeners"][0]["reality-config"] = json!({ + "dest":format!("localhost:{cover_port}"), "server-names":["localhost"], + "private-key":private, "short-id":["0123456789abcdef"] + }); + } else { + let proxy = &mut config["proxies"][0]; + proxy["tls"] = json!(true); + proxy["servername"] = json!("localhost"); + proxy["client-fingerprint"] = json!("chrome"); + proxy["reality-opts"] = json!({"public-key":public, "short-id":"0123456789abcdef"}); + } } } config @@ -797,6 +961,10 @@ async fn direct_case(peer: Kind, peer_client: bool, transport: Transport, mode: ); } } + // WebSocket and Mihomo's masked native wrapper can close both transport halves. + tls.get_mut().1.send_close_notify(); + tls.flush().await?; + assert_eq!(tls.read(&mut [0]).await?, 0, "data after TLS close"); } Ok::<_, io::Error>(()) }) @@ -805,14 +973,17 @@ async fn direct_case(peer: Kind, peer_client: bool, transport: Transport, mode: panic!("{result:?}\n{}", pair_logs(&client, &server)); } let shoes = if peer_client { &server } else { &client }; - assert_eq!( - shoes - .logs() - .contains("VISION WRITE: Switching to direct copy mode"), - !tls12, - "{}", - shoes.logs() - ); + shoes.assert_handshakes(&["fresh", "resumed"]); + for direction in ["READ", "WRITE"] { + assert_eq!( + shoes.logs().contains(&format!( + "VISION {direction}: Switching to direct copy mode" + )), + !tls12, + "{}", + shoes.logs() + ); + } drop((client, server)); tap_task.stop().await; target_task.stop().await; @@ -1014,6 +1185,9 @@ async fn wrong_uuid_and_key_order_have_no_destination_side_effects() { [0]["id"] } Kind::Mihomo => &mut config["proxies"][0]["uuid"], + Kind::SingBox => { + unreachable!("sing-box does not implement native encryption") + } }; *user = json!("c85798ef-e9dc-46a4-9a87-8da4499d36d0"); } else { diff --git a/tests/cases/optional_peer/vless_encryption/chained.rs b/tests/cases/optional_peer/vless_encryption/chained.rs new file mode 100644 index 00000000..7a283509 --- /dev/null +++ b/tests/cases/optional_peer/vless_encryption/chained.rs @@ -0,0 +1,99 @@ +use super::*; + +#[tokio::test] +async fn tls_and_anytls_consumers_survive_backpressure_and_cancelled_streams() { + let with_padding_gap = |setting: String| { + let gapped = setting.replace(".100-35-35.", ".100-35-35.100-25-25.100-35-35."); + assert_ne!(gapped, setting, "fixture padding token was not found"); + gapped + }; + for mode in ["native", "xorpub", "random"] { + for vision in [false, true] { + for anytls in [false, true] { + eprintln!("chained: {mode}, vision={vision}, anytls={anytls}"); + let (encryption, decryption, _) = credentials(mode, &["x25519"]); + let encryption = with_padding_gap(encryption); + let decryption = with_padding_gap(decryption); + let carrier_port = free_port().await; + let carrier = start( + Kind::Shoes, + carrier_port, + server_config(Kind::Shoes, carrier_port, &decryption, vision), + ) + .await; + let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + let hop_port = free_port().await; + let hop_protocol = if anytls { + json!({"type":"anytls","users":[{"name":"test","password":"test-password"}]}) + } else { + json!({"type":"socks"}) + }; + let hop_config = json!([{ + "address":format!("0.0.0.0:{hop_port}"), + "protocol":{"type":"tls","default_target":{ + "cert":cert.cert.pem(),"key":cert.signing_key.serialize_pem(),"protocol":hop_protocol + }} + }]); + let hop = start(Kind::Shoes, hop_port, hop_config).await; + let proxy = free_port().await; + let mut config = + client_config(Kind::Shoes, proxy, carrier_port, &encryption, vision); + let vless_hop = config[0]["rules"][0]["client_chain"].take(); + let hop_client_protocol = if anytls { + json!({"type":"anytls","password":"test-password"}) + } else { + json!({"type":"socks"}) + }; + config[0]["rules"][0]["client_chain"] = json!([vless_hop, { + "address":format!("127.0.0.1:{hop_port}"), + "protocol":{"type":"tls","sni_hostname":"localhost","verify":false,"protocol":hop_client_protocol} + }]); + let client = start(Kind::Shoes, proxy, config).await; + let (target, _, mut target_task) = echo().await; + let result = timeout(Duration::from_secs(30), async { + // Keep one stream alive while abandoning a different stream with unread data. + let mut survivor = socks_connect(proxy, target).await?; + let mut cancelled = socks_connect(proxy, target).await?; + cancelled.write_all(&vec![31; 128 * 1024]).await?; + drop(cancelled); + let payload: Vec<_> = + (0..2 * 1024 * 1024 + 17).map(|n| (n % 251) as u8).collect(); + let (mut read, mut write) = survivor.split(); + tokio::try_join!( + async { + for chunk in payload.chunks(16384) { + write.write_all(chunk).await?; + } + write.flush().await + }, + async { + let mut received = vec![0; payload.len()]; + for chunk in received.chunks_mut(4096) { + read.read_exact(chunk).await?; + tokio::time::sleep(Duration::from_millis(1)).await; + } + assert_eq!(received, payload); + Ok::<_, io::Error>(()) + } + )?; + drop(survivor); + exchange(proxy, target, b"new stream after cancellation").await + }) + .await; + assert!( + matches!(result, Ok(Ok(()))), + "{result:?}\n{}\nhop: {}", + pair_logs(&client, &carrier), + hop.logs() + ); + assert!( + carrier + .logs() + .contains("VLESS encryption server: fresh handshake") + ); + drop((client, hop, carrier)); + target_task.stop().await; + } + } + } +} diff --git a/tests/cases/optional_peer/vless_encryption/composition.rs b/tests/cases/optional_peer/vless_encryption/composition.rs new file mode 100644 index 00000000..0407f0bc --- /dev/null +++ b/tests/cases/optional_peer/vless_encryption/composition.rs @@ -0,0 +1,309 @@ +use super::*; +use shoes_test_support::{ + socks5::Socks5UdpAssociation, test_servers::start_udp_echo_server_with_suffix, +}; + +pub(super) fn wrap_xhttp( + kind: Kind, + mut config: Value, + server: bool, + (version, mode): (&str, &str), + cert: &str, + key: &str, + cover: u16, +) -> Value { + let carrier = match (version, kind) { + ("reality", _) => Some(Transport::Reality), + ("plain" | "h2c", _) => None, + ("h3", Kind::Shoes) => Some(Transport::Quic), + _ => Some(Transport::Tls), + }; + if let Some(carrier) = carrier { + config = wrap(kind, config, server, carrier, cert, key, cover); + } + let alpn = match version { + "plain" | "h1" => "http/1.1", + "h3" => "h3", + _ => "h2", + }; + match kind { + Kind::SingBox => unreachable!("sing-box does not implement XHTTP"), + Kind::Shoes => { + let mut protocol = shoes_protocol_mut(&mut config, server); + if protocol["type"] == "tls" { + protocol = if !server { + &mut protocol["protocol"] + } else if version == "reality" { + &mut protocol["reality_targets"]["localhost"]["protocol"] + } else { + &mut protocol["default_target"]["protocol"] + }; + } else if protocol["type"] == "reality" { + protocol = &mut protocol["protocol"]; + } + let inner = protocol.take(); + let http_version = match version { + "plain" => "h1", + "h2c" => "h2", + "reality" => "auto", + other => other, + }; + *protocol = json!({"type":"xhttp", "path":"/native", "host":"localhost", "mode":mode, + "http_version":http_version, + "sc_max_each_post_bytes":1024, "protocol":inner}); + } + Kind::Xray => { + let stream = &mut xray_endpoint_mut(&mut config, server)["streamSettings"]; + if stream.is_null() { + *stream = json!({"security":"none"}); + } + stream["network"] = json!("xhttp"); + stream["xhttpSettings"] = json!({"path":"/native", "host":"localhost", "mode":mode, + "extra":{"scMaxEachPostBytes":1024, "scMinPostsIntervalMs":0}}); + if stream["security"] == "tls" { + stream["tlsSettings"]["alpn"] = json!([alpn]); + } + if server && version == "h3" { + let port = config["inbounds"][0]["port"].clone(); + config["inbounds"].as_array_mut().unwrap().push(json!({ + "listen":"0.0.0.0","port":port,"protocol":"socks","settings":{"auth":"noauth"} + })); + } + } + Kind::Mihomo => { + let options = json!({"path":"/native", "host":"localhost", "mode":mode, + "sc-max-each-post-bytes":"1024", "sc-min-posts-interval-ms":"1"}); + if server { + assert_ne!(version, "h3", "Mihomo does not serve XHTTP over QUIC"); + config["listeners"][0]["xhttp-config"] = options; + if matches!(version, "plain" | "h2c") { + config["listeners"][0]["allow-insecure"] = json!(true); + } + } else { + config["proxies"][0]["network"] = json!("xhttp"); + config["proxies"][0]["alpn"] = json!([alpn]); + config["proxies"][0]["xhttp-opts"] = options; + } + } + } + config +} + +async fn wrapped_case( + peer: Kind, + peer_client: bool, + transport: Transport, + encrypted: bool, + vision: bool, +) { + eprintln!( + "composition: {peer:?} client={peer_client} {transport:?} encrypted={encrypted} vision={vision}" + ); + let (encryption, decryption) = if encrypted { + let (encryption, decryption, _) = credentials("random", &["mlkem768"]); + (encryption, decryption) + } else { + ("none".into(), "none".into()) + }; + let (server_kind, client_kind) = peer_roles(peer, peer_client); + let (cover, connector, mut cover_task) = tls_echo(false).await; + let (target, _, mut target_task) = echo().await; + let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + let (cert, key) = (cert.cert.pem(), cert.signing_key.serialize_pem()); + let server_port = free_port().await; + let mut server_settings = server_config(server_kind, server_port, &decryption, vision); + if server_kind == Kind::Mihomo && !encrypted { + server_settings["listeners"][0]["allow-insecure"] = json!(true); + } + let server = start( + server_kind, + server_port, + wrap( + server_kind, + server_settings, + true, + transport, + &cert, + &key, + cover, + ), + ) + .await; + let client_port = free_port().await; + let mut client_settings = + client_config(client_kind, client_port, server_port, &encryption, vision); + if client_kind == Kind::Mihomo { + client_settings["proxies"][0]["packet-encoding"] = json!("xudp"); + } + let client = start( + client_kind, + client_port, + wrap( + client_kind, + client_settings, + false, + transport, + &cert, + &key, + cover, + ), + ) + .await; + let data: Vec<_> = (0..32769).map(|n| (n % 251) as u8).collect(); + for _ in 0..2 { + if vision { + let result = timeout(LIMIT, async { + let socket = socks_connect(client_port, cover).await?; + let mut stream = connector + .connect("localhost".try_into().unwrap(), socket) + .await?; + for payload in [&[42; 64][..], &[43; 64], &data] { + let (mut r, mut w) = tokio::io::split(&mut stream); + let mut reply = vec![0; payload.len()]; + tokio::try_join!(w.write_all(payload), r.read_exact(&mut reply))?; + assert_eq!(reply, payload); + } + // Close inner TLS without requiring WebSocket to support a TCP half-close. + stream.get_mut().1.send_close_notify(); + stream.flush().await?; + assert_eq!(stream.read(&mut [0]).await?, 0, "data after TLS close"); + Ok::<_, io::Error>(()) + }) + .await; + assert!( + matches!(result, Ok(Ok(()))), + "{result:?}\n{}", + pair_logs(&client, &server) + ); + } else { + exchange(client_port, target, &data) + .await + .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server))); + } + } + let shoes = if peer_client { &server } else { &client }; + if encrypted { + shoes.assert_handshakes(&["fresh", "resumed"]); + } + if vision { + for direction in ["READ", "WRITE"] { + assert_eq!( + shoes + .logs() + .matches(&format!( + "VISION {direction}: Switching to direct copy mode" + )) + .count(), + 2, + "{}", + shoes.logs() + ); + } + } + // Force UDP frames across HTTP POST and native-record boundaries. + let udp = start_udp_echo_server_with_suffix("0.0.0.0", 0, b"reply") + .await + .unwrap(); + let destination = (std::net::Ipv4Addr::LOCALHOST, udp.local_addr().port()).into(); + for _ in 0..2 { + let association = timeout( + LIMIT, + Socks5UdpAssociation::connect("127.0.0.1", client_port), + ) + .await + .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server))) + .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server))); + let payload = vec![37; 4096]; + let reply = association + .send_to(destination, &payload) + .await + .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server))); + assert_eq!(reply, [payload, b"reply".to_vec()].concat()); + } + drop((client, server)); + cover_task.stop().await; + target_task.stop().await; +} + +#[tokio::test] +async fn encrypted_xhttp_xray_client_resumes_and_carries_tcp_udp_and_vision() { + xhttp_matrix(Kind::Xray, true).await; +} + +#[tokio::test] +async fn encrypted_xhttp_xray_server_resumes_and_carries_tcp_udp_and_vision() { + xhttp_matrix(Kind::Xray, false).await; +} + +#[tokio::test] +async fn encrypted_xhttp_mihomo_client_resumes_and_carries_tcp_udp_and_vision() { + xhttp_matrix(Kind::Mihomo, true).await; +} + +#[tokio::test] +async fn encrypted_xhttp_mihomo_server_resumes_and_carries_tcp_udp_and_vision() { + xhttp_matrix(Kind::Mihomo, false).await; +} + +async fn xhttp_matrix(peer: Kind, peer_client: bool) { + for version in ["plain", "h1", "h2c", "h2", "h3", "reality"] { + // Xray's client selects HTTP/1.1 without TLS; its server accepts h2c. + if peer == Kind::Xray && peer_client && version == "h2c" { + continue; + } + if peer == Kind::Mihomo && !peer_client && version == "h3" { + continue; + } + for mode in ["packet-up", "stream-up", "stream-one", "auto"] { + if mode == "auto" && !matches!(version, "h2" | "reality") { + continue; + } + for vision in [false, true] { + wrapped_case( + peer, + peer_client, + Transport::XhttpMode { version, mode }, + true, + vision, + ) + .await; + } + } + } +} + +#[tokio::test] +async fn encrypted_non_vision_and_wss_resume_through_wrappers() { + for peer in [Kind::Xray, Kind::Mihomo] { + for client in [false, true] { + for transport in [ + Transport::Tls, + Transport::Websocket, + Transport::Wss, + Transport::Reality, + ] { + wrapped_case(peer, client, transport, true, false).await; + } + wrapped_case(peer, client, Transport::Wss, true, true).await; + } + } + for vision in [false, true] { + wrapped_case(Kind::Shoes, false, Transport::Quic, true, vision).await; + } +} + +#[tokio::test] +async fn legacy_vless_websocket_and_mihomo_transports_carry_tcp_and_udp() { + for peer in [Kind::Xray, Kind::Mihomo] { + for client in [false, true] { + for transport in [Transport::Websocket, Transport::Wss] { + wrapped_case(peer, client, transport, false, false).await; + } + } + } + for client in [false, true] { + for transport in [Transport::Raw, Transport::Tls, Transport::Reality] { + wrapped_case(Kind::Mihomo, client, transport, false, false).await; + } + } +} diff --git a/tests/cases/optional_peer/vless_encryption/legacy.rs b/tests/cases/optional_peer/vless_encryption/legacy.rs index ad7fe4fe..5e7b3214 100644 --- a/tests/cases/optional_peer/vless_encryption/legacy.rs +++ b/tests/cases/optional_peer/vless_encryption/legacy.rs @@ -1,5 +1,244 @@ use super::*; +pub(super) fn wrap_singbox( + mut config: Value, + server: bool, + transport: Transport, + cert: &str, + key: &str, + cover: u16, +) -> Value { + let tls = if matches!(transport, Transport::Reality) { + let (private, public) = reality_keys(); + if server { + json!({"enabled":true,"server_name":"localhost","reality":{"enabled":true, + "handshake":{"server":"localhost","server_port":cover},"private_key":private,"short_id":["0123456789abcdef"]}}) + } else { + json!({"enabled":true,"server_name":"localhost","utls":{"enabled":true,"fingerprint":"chrome"}, + "reality":{"enabled":true,"public_key":public,"short_id":"0123456789abcdef"}}) + } + } else { + assert!(matches!(transport, Transport::Tls)); + if server { + json!({"enabled":true,"certificate":cert.lines().collect::>(),"key":key.lines().collect::>()}) + } else { + json!({"enabled":true,"server_name":"localhost","insecure":true}) + } + }; + config[if server { "inbounds" } else { "outbounds" }][0]["tls"] = tls; + config +} + +fn enable_legacy_vision(config: &mut Value, server: bool, transport: Transport) { + let outer = shoes_protocol_mut(config, server); + let target = if !server { + outer + } else if matches!(transport, Transport::Tls) { + &mut outer["default_target"] + } else { + &mut outer["reality_targets"]["localhost"] + }; + target["vision"] = json!(true); +} + +async fn checked_transfer( + mut stream: S, + half_close_before_reply: bool, +) -> io::Result { + for _ in 0..2 { + stream.write_all(&[42; 64]).await?; + stream.flush().await?; + let mut reply = [0; 64]; + stream.read_exact(&mut reply).await?; + assert_eq!(reply, [42; 64]); + } + let data: Vec<_> = (0..1024 * 1024 + 17).map(|i| (i % 251) as u8).collect(); + let (mut r, mut w) = tokio::io::split(&mut stream); + let ((), output) = tokio::try_join!( + async { + w.write_all(&data).await?; + if half_close_before_reply { + w.shutdown().await?; + } + Ok::<_, io::Error>(()) + }, + async { + let mut output = Vec::new(); + let mut buf = [0; 8192]; + let mut next_pause = 65536; + while output.len() < data.len() { + let n = r.read(&mut buf).await?; + if n == 0 { + break; + } + output.extend_from_slice(&buf[..n]); + assert!(output.len() <= data.len(), "duplicate data before EOF"); + if output.len() >= next_pause { + next_pause += 65536; + tokio::time::sleep(Duration::from_millis(1)).await; + } + } + if half_close_before_reply { + assert_eq!(r.read(&mut buf).await?, 0, "data after expected reply"); + } + Ok::<_, io::Error>(output) + } + )?; + if output != data { + return Err(io::Error::other(format!( + "echo mismatch before EOF: received {} of {} bytes, first mismatch {:?}", + output.len(), + data.len(), + output.iter().zip(&data).position(|(a, b)| a != b), + ))); + } + drop((r, w)); + Ok(stream) +} + +#[tokio::test] +async fn legacy_peers_assert_direct_end_and_eof_over_tls_and_reality() { + for peer in [Kind::Xray, Kind::Mihomo, Kind::SingBox] { + for peer_client in [false, true] { + for transport in [Transport::Tls, Transport::Reality] { + for inner_tls12 in [None, Some(true), Some(false)] { + // sing-box's Vision wrapper has no CloseWrite. Clients close both sides on + // upload EOF; servers can reset TLS 1.2 replies when the target closes first. + // Upstream fix: https://github.com/SagerNet/sing-vmess/pull/16 + let half_close_before_reply = + peer != Kind::SingBox || (!peer_client && inner_tls12 != Some(true)); + decision_case( + peer, + peer_client, + transport, + inner_tls12, + half_close_before_reply, + ) + .await; + } + } + } + } +} + +#[tokio::test] +#[ignore = "pinned sing-box Vision client closes both directions on upload EOF"] +async fn singbox_vision_client_half_close_preserves_pending_reply() { + decision_case(Kind::SingBox, true, Transport::Tls, None, true).await; +} + +#[tokio::test] +#[ignore = "pinned sing-box Vision server can reset TLS 1.2 replies during upload half-close"] +async fn singbox_vision_server_half_close_preserves_tls12_reply() { + for transport in [Transport::Tls, Transport::Reality] { + decision_case(Kind::SingBox, false, transport, Some(true), true).await; + } +} + +async fn decision_case( + peer: Kind, + peer_client: bool, + transport: Transport, + inner_tls12: Option, + half_close_before_reply: bool, +) { + eprintln!( + "legacy decisions: {peer:?} client={peer_client} {transport:?} inner_tls12={inner_tls12:?}" + ); + let (server_kind, client_kind) = peer_roles(peer, peer_client); + let (cover, _, mut cover_task) = tls_echo(false).await; + let (target, connector, mut target_task) = if let Some(tls12) = inner_tls12 { + let (port, connector, task) = tls_echo(tls12).await; + (port, Some(connector), task) + } else { + let (port, _, task) = echo().await; + (port, None, task) + }; + let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + let (cert, key) = (cert.cert.pem(), cert.signing_key.serialize_pem()); + let server_port = free_port().await; + let client_port = free_port().await; + let mut server_settings = wrap( + server_kind, + server_config(server_kind, server_port, "none", server_kind != Kind::Shoes), + true, + transport, + &cert, + &key, + cover, + ); + let mut client_settings = wrap( + client_kind, + client_config( + client_kind, + client_port, + server_port, + "none", + client_kind != Kind::Shoes, + ), + false, + transport, + &cert, + &key, + cover, + ); + if server_kind == Kind::Shoes { + enable_legacy_vision(&mut server_settings, true, transport); + } else { + enable_legacy_vision(&mut client_settings, false, transport); + } + let server = start(server_kind, server_port, server_settings).await; + let client = start(client_kind, client_port, client_settings).await; + timeout(Duration::from_secs(20), async { + let socket = socks_connect(client_port, target).await?; + if let Some(connector) = connector { + let mut tls = checked_transfer( + connector + .connect("localhost".try_into().unwrap(), socket) + .await?, + half_close_before_reply, + ) + .await?; + if !half_close_before_reply { + tls.get_mut().1.send_close_notify(); + tls.flush().await?; + assert_eq!(tls.read(&mut [0]).await?, 0, "data after TLS close"); + } + } else { + let mut socket = checked_transfer(socket, half_close_before_reply).await?; + if !half_close_before_reply { + socket.shutdown().await?; + assert_eq!(socket.read(&mut [0]).await?, 0, "data after expected reply"); + } + } + Ok::<_, io::Error>(()) + }) + .await + .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server))) + .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server))); + let shoes = if peer_client { &server } else { &client }; + let direct = inner_tls12 == Some(false); + for direction in ["READ", "WRITE"] { + let logs = shoes.logs(); + assert_eq!( + logs.contains(&format!( + "VISION {direction}: Switching to direct copy mode" + )), + direct, + "{logs}" + ); + if !direct { + assert!( + logs.contains(&format!("VISION {direction}: Switching to Tls mode")), + "{logs}" + ); + } + } + drop((client, server)); + cover_task.stop().await; + target_task.stop().await; +} + #[derive(Debug, PartialEq, Eq)] struct LegacyObservation { eof: bool, diff --git a/tests/cases/optional_peer/vless_encryption/lifecycle.rs b/tests/cases/optional_peer/vless_encryption/lifecycle.rs new file mode 100644 index 00000000..d2a382b7 --- /dev/null +++ b/tests/cases/optional_peer/vless_encryption/lifecycle.rs @@ -0,0 +1,216 @@ +use super::*; + +#[cfg(unix)] +async fn reload(peer: &Peer, config: &Value) { + let ready = peer.logs().matches("Servers ready").count(); + let path = peer.directory.path().join("config.json"); + std::fs::write(path, serde_json::to_vec(config).unwrap()).unwrap(); + assert_eq!( + unsafe { libc::kill(peer.process.id() as i32, libc::SIGHUP) }, + 0 + ); + timeout(LIMIT, async { + while peer.logs().matches("Servers ready").count() == ready { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap_or_else(|e| panic!("reload: {e}\n{}", peer.logs())); +} + +async fn roundtrip(stream: &mut TcpStream, payload: &[u8]) { + timeout(LIMIT, async { + stream.write_all(payload).await.unwrap(); + let mut reply = vec![0; payload.len()]; + stream.read_exact(&mut reply).await.unwrap(); + assert_eq!(reply, payload); + }) + .await + .unwrap(); +} + +#[cfg(unix)] +#[tokio::test] +async fn live_reload_resets_tickets_preserves_active_streams_and_rotates_keys() { + for peer in [Kind::Xray, Kind::Mihomo] { + for mode in ["native", "xorpub", "random"] { + let (encryption, decryption, _) = credentials(mode, &["x25519"]); + let (target, count, mut target_task) = echo().await; + let port = free_port().await; + let mut config = server_config(Kind::Shoes, port, &decryption, false); + let server = start(Kind::Shoes, port, config.clone()).await; + let client_port = free_port().await; + let client = start( + peer, + client_port, + client_config(peer, client_port, port, &encryption, false), + ) + .await; + let mut active = socks_connect(client_port, target).await.unwrap(); + roundtrip(&mut active, b"old generation").await; + exchange(client_port, target, b"warm").await.unwrap(); + server.assert_handshakes(&["fresh", "resumed"]); + + config[0]["protocol"]["decryption"] = json!(decryption.replace(".600s.", ".0s.")); + reload(&server, &config).await; + roundtrip(&mut active, b"still live after reload").await; + assert!( + exchange(client_port, target, b"stale must not arrive") + .await + .is_err() + ); + assert_eq!(count.load(Ordering::SeqCst), 2); + for _ in 0..2 { + exchange(client_port, target, b"fresh-only recovery") + .await + .unwrap(); + } + server.assert_handshakes(&["fresh", "resumed", "fresh", "fresh"]); + + let (next_encryption, next_decryption, _) = credentials(mode, &["x25519"]); + config[0]["protocol"]["decryption"] = json!(next_decryption); + reload(&server, &config).await; + roundtrip(&mut active, b"old stream survives key rotation").await; + assert!( + exchange(client_port, target, b"old key must fail") + .await + .is_err() + ); + assert_eq!(count.load(Ordering::SeqCst), 4); + drop(client); + let client = start( + peer, + client_port, + client_config(peer, client_port, port, &next_encryption, false), + ) + .await; + for _ in 0..2 { + exchange(client_port, target, b"new key").await.unwrap(); + } + server.assert_handshakes(&["fresh", "resumed", "fresh", "fresh", "fresh", "resumed"]); + drop((active, client, server)); + target_task.stop().await; + } + } +} + +#[tokio::test] +async fn expired_tickets_use_a_fresh_handshake_without_losing_the_request() { + for peer in [Kind::Xray, Kind::Mihomo] { + for peer_client in [false, true] { + let (server_kind, client_kind) = peer_roles(peer, peer_client); + let (encryption, decryption, _) = credentials("random", &["x25519"]); + let decryption = decryption.replace(".600s.", ".5-5s."); + let (target, count, mut task) = echo().await; + let port = free_port().await; + let server = start( + server_kind, + port, + server_config(server_kind, port, &decryption, false), + ) + .await; + let proxy = free_port().await; + let client = start( + client_kind, + proxy, + client_config(client_kind, proxy, port, &encryption, false), + ) + .await; + for _ in 0..2 { + exchange(proxy, target, b"warm before expiry") + .await + .unwrap(); + } + let shoes = if peer_client { &server } else { &client }; + shoes.assert_handshakes(&["fresh", "resumed"]); + tokio::time::sleep(Duration::from_millis(5200)).await; + exchange(proxy, target, b"fresh after expiry") + .await + .unwrap(); + shoes.assert_handshakes(&["fresh", "resumed", "fresh"]); + assert_eq!(count.load(Ordering::SeqCst), 3); + drop((client, server)); + task.stop().await; + } + } +} + +#[tokio::test] +async fn shoes_client_rejects_stale_peer_ticket_without_replaying_application_data() { + for peer in [Kind::Xray, Kind::Mihomo] { + for mode in ["native", "xorpub", "random"] { + let (encryption, decryption, _) = credentials(mode, &["x25519"]); + let (target, count, mut task) = echo().await; + let port = free_port().await; + let config = server_config(peer, port, &decryption, false); + let server = start(peer, port, config.clone()).await; + let proxy = free_port().await; + let client = start( + Kind::Shoes, + proxy, + client_config(Kind::Shoes, proxy, port, &encryption, false), + ) + .await; + for _ in 0..2 { + exchange(proxy, target, b"warm").await.unwrap(); + } + client.assert_handshakes(&["fresh", "resumed"]); + drop(server); + let server = start(peer, port, config).await; + assert!( + exchange(proxy, target, b"do not retry this request") + .await + .is_err() + ); + assert_eq!(count.load(Ordering::SeqCst), 2); + exchange(proxy, target, b"independent recovery") + .await + .unwrap(); + assert_eq!(count.load(Ordering::SeqCst), 3); + client.assert_handshakes(&["fresh", "resumed", "resumed", "fresh"]); + drop((client, server)); + task.stop().await; + } + } +} + +#[tokio::test] +async fn live_chacha_peer_clients_complete_fresh_and_resumed_handshakes() { + for peer in [Kind::Xray, Kind::Mihomo] { + for mode in ["native", "random"] { + let (encryption, decryption, _) = credentials(mode, &["mlkem768"]); + let (target, _, mut task) = echo().await; + let port = free_port().await; + let server = start( + Kind::Shoes, + port, + server_config(Kind::Shoes, port, &decryption, false), + ) + .await; + let proxy = free_port().await; + let client = start_options( + peer, + proxy, + client_config(peer, proxy, port, &encryption, false), + peer.binary(), + Some("cpu.aes=off,cpu.pclmulqdq=off"), + ) + .await; + for _ in 0..2 { + exchange(proxy, target, &vec![42; 32768]).await.unwrap(); + } + server.assert_handshakes(&["fresh", "resumed"]); + assert_eq!( + server + .logs() + .matches("VLESS encryption server cipher: ChaCha20Poly1305") + .count(), + 2, + "{}", + server.logs() + ); + drop((client, server)); + task.stop().await; + } + } +} diff --git a/tests/cases/optional_peer/wireguard_interop.rs b/tests/cases/optional_peer/wireguard_interop.rs index d056de28..0bc188fd 100644 --- a/tests/cases/optional_peer/wireguard_interop.rs +++ b/tests/cases/optional_peer/wireguard_interop.rs @@ -29,7 +29,7 @@ fn singbox_server(port: u16) -> Value { } #[tokio::test] -#[ignore = "Requires Xray 26.10.10; set SHOES_TEST_XRAY_BIN"] +#[ignore = "Requires Xray 26.10.17; set SHOES_TEST_XRAY_BIN"] async fn wireguard_xray_both_roles_tcp_udp_ipv4_ipv6() -> io::Result<()> { tokio::time::timeout(Duration::from_secs(90), async { for shoes_is_server in [false, true] { diff --git a/tests/fixtures/vless_encryption/README.md b/tests/fixtures/vless_encryption/README.md index 51ac878d..d3bfbf66 100644 --- a/tests/fixtures/vless_encryption/README.md +++ b/tests/fixtures/vless_encryption/README.md @@ -37,15 +37,61 @@ GOMAXPROCS=1 go run -p 1 /path/to/shoes/tests/fixtures/vless_encryption/generate --verify-rust "$HOME/vless-rust-vectors.json" ``` -The native peer suite uses `XRAY_BINARY`, `MIHOMO_BINARY` and optionally -`SHOES_TEST_SHOES_BIN`. Missing requested peers fail rather than skip. Its wire -tap distinguishes fresh from resumed handshakes and checks raw DIRECT lengths. +The peer suite uses `XRAY_BINARY`, `MIHOMO_BINARY`, `SHOES_TEST_SING_BOX_BIN`, and +optionally `SHOES_TEST_SHOES_BIN`. CI pins Xray 26.10.17 +(`92ccc939aaf4f8965f369cf0b142e005aeaa11d5`), including its Vision TLS-buffer +compatibility fix. The committed vectors retain their original provenance above. +Missing requested peers fail rather than skip. The wire tap distinguishes fresh +from resumed handshakes and checks raw DIRECT lengths. Successful transfers plus +secret-free debug diagnostics verify resumption inside opaque carriers. Xray's command-2 replies are limited to 8190 bytes; its XUDP writer limits payloads to 7526 bytes. Peer UDP tests use that intersection. Separate Shoes tests cover larger XUDP frames and preservation of the existing empty-DATA and control-frame semantics. Sing-box has no native encryption support and is used only by the legacy TLS/REALITY regression suites. +The enabled matrix covers: + +- Native TCP and UDP over TLS, WebSocket, WSS, REALITY, and Shoes raw QUIC. +- Encrypted XHTTP with and without Vision: packet-up, stream-up, and stream-one + over H1, H2C, H2, H3, and REALITY; auto mode over H2 and REALITY. Xray cannot + initiate H2C, and Mihomo cannot serve H3; those directions are excluded. +- Legacy plain VLESS over WS/WSS and Mihomo raw/TLS/REALITY. Legacy Vision uses + Xray, Mihomo, and sing-box in both roles, asserting DIRECT for inner TLS 1.3, + END for plain/TLS 1.2 traffic, exact payloads under slow reads, and remote EOF. + Clients half-close before receiving the full reply except in the sing-box cases + below. + The pinned sing-box Vision client closes both directions on upload EOF, so its + enabled cases finish receiving before closing. The ignored + `singbox_vision_client_half_close_preserves_pending_reply` retains that failing + peer reproducer. The sing-box Vision server can also reset inner TLS 1.2 replies + when the target closes before upload EOF arrives. Only those server cases, over + TLS and REALITY, receive the full payload before sending inner `close_notify` + and requiring clean EOF. The ignored + `singbox_vision_server_half_close_preserves_tls12_reply` retains the intermittent + server reproducer. Remove this server workaround when the pinned peer includes + the Vision half-close fix in + [SagerNet/sing-vmess#16](https://github.com/SagerNet/sing-vmess/pull/16). + Plain and TLS 1.3 server cases still assert early half-close. + All native-Vision peer cases close inner TLS gracefully without + requiring transport-level half-close support: WebSocket and Mihomo's masked + native wrapper can close both transport halves. Native TCP half-close and a + delayed reverse reply are asserted separately in the library/splice tests. +- Live server reload, zero ticket lifetime, key rotation, expiry, peer restart, + and recovery on a new request without replaying failed application data. +- Live ChaCha20-Poly1305 from peer clients with Go AES acceleration disabled. +- Actual TLS/SOCKS and TLS/AnyTLS consumers chained over native encryption, + including slow full-duplex reads and cancellation of another stream. + +The library suite complements the process matrix with native fresh/resumed +Vision decisions, fragmented and malformed records, the maximum native record +across Vision's 8 KiB buffer, real Linux splice transfers and masked/wrapped +negative controls, concurrent duplicate flights across bind IPs, SNI/WS ticket +isolation, IPv4/IPv6/domain destinations, empty UDP replies, and native-Vision +H2MUX rejection. IPv6 address cases run when IPv6 loopback is available; +IPv4 and domain cases always run without relying on dual-stack sockets. +These tests also run without optional peer binaries. + The ignored `legacy_baseline_matches_backpressure_and_half_close` test compares both Shoes roles against Xray over TLS, including slow reads and half-close after DIRECT. It verifies payload integrity, observes remote EOF within five diff --git a/tests/support/src/xray.rs b/tests/support/src/xray.rs index 94b669ed..33c73547 100644 --- a/tests/support/src/xray.rs +++ b/tests/support/src/xray.rs @@ -11,7 +11,7 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use crate::process::ProcessGuard; -pub const VERSION: &str = "26.10.10"; +pub const VERSION: &str = "26.10.17"; pub struct Peer { _process: ProcessGuard,