From aae1945e5cd4463b436e07ea9b3d5259efe819b7 Mon Sep 17 00:00:00 2001 From: cfal Date: Sat, 17 Oct 2026 14:23:56 -0400 Subject: [PATCH] Fix Vision DIRECT decisions under backpressure (#187) Preserve recognized TLS framing when backpressure requires resubmitting an unaccepted write tail, preventing eligible Vision connections from incorrectly falling back to END. Restrict DIRECT to supported TLS 1.3 cipher suites and keep ServerHello parsing within its declared handshake and extension boundaries. Add regression coverage for TLS and REALITY, fragmented and coalesced records, malformed handshakes, and exact-byte preservation across DIRECT and END transitions. --- src/shadow_tls/shadow_tls_server_handler.rs | 12 + src/shadow_tls/tests/server_hello.rs | 108 ++++ src/vless/tls_fuzzy_deframer.rs | 60 +- src/vless/vision_filter.rs | 19 +- src/vless/vision_stream.rs | 7 +- src/vless/vision_stream/tests/decisions.rs | 630 ++++++++++++++++++++ 6 files changed, 819 insertions(+), 17 deletions(-) create mode 100644 src/shadow_tls/tests/server_hello.rs create mode 100644 src/vless/vision_stream/tests/decisions.rs diff --git a/src/shadow_tls/shadow_tls_server_handler.rs b/src/shadow_tls/shadow_tls_server_handler.rs index ecb1feba..99d60774 100644 --- a/src/shadow_tls/shadow_tls_server_handler.rs +++ b/src/shadow_tls/shadow_tls_server_handler.rs @@ -575,6 +575,8 @@ pub fn parse_server_hello(server_hello_frame: &[u8]) -> std::io::Result std::io::Result) -> Vec { + let mut body = vec![3, 3]; + body.extend_from_slice(&[42; 32]); + body.push(session_id_len); + body.extend(std::iter::repeat_n(7, usize::from(session_id_len))); + body.extend_from_slice(&[0x13, 0x01, 0]); + if let Some(extensions) = extensions { + body.extend_from_slice(&u16::try_from(extensions.len()).unwrap().to_be_bytes()); + body.extend_from_slice(extensions); + } + let mut frame = vec![0x16, 3, 3]; + frame.extend_from_slice(&u16::try_from(4 + body.len()).unwrap().to_be_bytes()); + frame.push(2); + frame.extend_from_slice(&u32::try_from(body.len()).unwrap().to_be_bytes()[1..]); + frame.extend_from_slice(&body); + frame +} + +#[test] +fn parses_tls12_and_tls13_with_variable_session_ids() { + for session_id_len in [0, 1, 32] { + for (extensions, expected_tls13) in [ + (None, false), + (Some(b"".as_slice()), false), + (Some(TLS13_VERSION), true), + ] { + let frame = server_hello(session_id_len, extensions); + let parsed = parse_server_hello(&frame).unwrap(); + assert_eq!(parsed.server_random, [42; 32]); + assert_eq!(parsed.session_id_len, session_id_len); + assert_eq!(parsed.cipher_suite, 0x1301); + assert_eq!(parsed.is_tls13, expected_tls13); + assert_eq!( + parse_validated_server_hello(&frame).is_ok(), + parsed.is_tls13 && session_id_len == 32 + ); + } + } +} + +#[test] +fn coalesced_handshakes_do_not_extend_server_hello() { + for (extensions, expected_tls13) in [ + (None, false), + (Some(b"".as_slice()), false), + (Some(TLS13_VERSION), true), + ] { + let mut frame = server_hello(32, extensions); + // A second handshake containing a version-like byte sequence must be ignored. + frame.extend_from_slice(&[11, 0, 0, 6]); + frame.extend_from_slice(TLS13_VERSION); + let record_len = u16::try_from(frame.len() - 5).unwrap(); + frame[3..5].copy_from_slice(&record_len.to_be_bytes()); + let parsed = parse_server_hello(&frame).unwrap(); + assert_eq!(parsed.is_tls13, expected_tls13); + } +} + +#[test] +fn declared_handshake_body_bounds_every_field() { + for session_id_len in [0, 1, 32] { + let valid = server_hello(session_id_len, Some(TLS13_VERSION)); + let minimum_body_len = 38 + usize::from(session_id_len); + for body_len in 0..valid.len() - 9 { + let mut frame = valid.clone(); + frame[6..9].copy_from_slice(&u32::try_from(body_len).unwrap().to_be_bytes()[1..]); + let parsed = parse_server_hello(&frame); + if body_len == minimum_body_len { + // TLS 1.2 permits a ServerHello with no extensions vector. + assert!(!parsed.unwrap().is_tls13); + } else { + assert!( + parsed.is_err(), + "session_id_len={session_id_len}, body_len={body_len}" + ); + } + } + for frame_len in 0..valid.len() { + assert!(parse_server_hello(&valid[..frame_len]).is_err()); + } + } +} + +#[test] +fn extensions_must_fill_the_declared_body_and_contain_complete_entries() { + let valid = server_hello(32, Some(TLS13_VERSION)); + for extension_len in 0..=TLS13_VERSION.len() + 1 { + let mut frame = valid.clone(); + frame[79..81].copy_from_slice(&u16::try_from(extension_len).unwrap().to_be_bytes()); + assert_eq!( + parse_server_hello(&frame).is_ok(), + extension_len == TLS13_VERSION.len(), + "extension_len={extension_len}" + ); + } + for extensions in [ + b"\x00\x2b\x00".as_slice(), + b"\x00\x2b\x00\x01\x03".as_slice(), + b"\x00\x2b\x00\x02\x03".as_slice(), + b"\xfa\xfa\x00\x02\x03".as_slice(), + ] { + assert!(parse_server_hello(&server_hello(32, Some(extensions))).is_err()); + } +} diff --git a/src/vless/tls_fuzzy_deframer.rs b/src/vless/tls_fuzzy_deframer.rs index 1374485f..76b44cf9 100644 --- a/src/vless/tls_fuzzy_deframer.rs +++ b/src/vless/tls_fuzzy_deframer.rs @@ -197,10 +197,15 @@ impl FuzzyTlsDeframer { } } - /// Clear all buffered data and reset to searching state - pub fn clear(&mut self) { + /// Discards buffered bytes for resubmission without forgetting recognized TLS framing. + pub fn discard_buffered(&mut self) { self.search_buffer.clear(); self.inner.clear(); + } + + /// Clear all buffered data and reset to searching state + pub fn clear(&mut self) { + self.discard_buffered(); self.state = FuzzyState::Searching; self.total_prefix_bytes = 0; } @@ -514,6 +519,57 @@ mod tests { assert_eq!(deframer.state, FuzzyState::Searching); } + #[test] + fn discard_buffered_preserves_tls_framing_at_a_record_boundary() { + let hello = make_tls_record(0x16, b"hello"); + let application = make_tls_record(0x17, b"application"); + for tail_len in 0..=application.len() { + let mut deframer = FuzzyTlsDeframer::new(); + deframer.feed(&hello); + assert!(matches!( + deframer.next_record().unwrap(), + DeframeResult::TlsRecord(_) + )); + deframer.feed(&application[..tail_len]); + if tail_len < application.len() { + assert!(matches!( + deframer.next_record().unwrap(), + DeframeResult::NeedData + )); + } + deframer.discard_buffered(); + assert_eq!(deframer.pending_bytes(), 0); + deframer.feed(&application); + match deframer.next_record().unwrap() { + DeframeResult::TlsRecord(bytes) => assert_eq!(bytes.as_ref(), application), + result => panic!("lost TLS framing: {result:?}"), + } + assert_eq!(deframer.pending_bytes(), 0); + assert!(matches!( + deframer.next_record().unwrap(), + DeframeResult::NeedData + )); + } + } + + #[test] + fn discard_buffered_does_not_promote_searching_to_tls() { + let mut deframer = FuzzyTlsDeframer::new(); + deframer.feed(&[0x16, 3]); + assert!(matches!( + deframer.next_record().unwrap(), + DeframeResult::NeedData + )); + deframer.discard_buffered(); + assert_eq!(deframer.pending_bytes(), 0); + let application = make_tls_record(0x17, b"application"); + deframer.feed(&application); + match deframer.next_record().unwrap() { + DeframeResult::UnknownPrefix(bytes) => assert_eq!(bytes.as_ref(), application), + result => panic!("unexpected TLS recognition: {result:?}"), + } + } + #[test] fn test_false_tls_pattern_continues_search() { let mut deframer = FuzzyTlsDeframer::new(); diff --git a/src/vless/vision_filter.rs b/src/vless/vision_filter.rs index fe09eca3..ae68e207 100644 --- a/src/vless/vision_filter.rs +++ b/src/vless/vision_filter.rs @@ -7,9 +7,6 @@ const TLS_CONTENT_TYPE_HANDSHAKE: u8 = 0x16; const TLS_HANDSHAKE_TYPE_CLIENT_HELLO: u8 = 0x01; const TLS_HANDSHAKE_TYPE_SERVER_HELLO: u8 = 0x02; -// TLS ciphers -const TLS13_CIPHER_AES_128_CCM_8_SHA256: u16 = 0x1305; - /// VISION protocol filter state /// /// This structure maintains only the state needed for TLS record filtering @@ -126,20 +123,18 @@ impl VisionFilter { if parsed.is_tls13 { log::debug!("VISION: Detected TLS 1.3 via supported_versions extension"); - // Check if cipher is supported for XTLS - // Following Xray-core: all TLS 1.3 ciphers EXCEPT 0x1305 support XTLS - if parsed.cipher_suite == TLS13_CIPHER_AES_128_CCM_8_SHA256 { - log::warn!( - "VISION: TLS 1.3 detected but cipher 0x{:04x} (TLS_AES_128_CCM_8_SHA256) not supported for XTLS", - parsed.cipher_suite - ); - } else { - // Assume all other TLS 1.3 ciphers support XTLS + // Xray and Mihomo allow GCM, ChaCha20 and CCM, but not CCM_8 or unknown suites. + if matches!(parsed.cipher_suite, 0x1301..=0x1304) { self.supports_xtls = true; log::debug!( "VISION: TLS 1.3 with supported cipher 0x{:04x} - XTLS enabled", parsed.cipher_suite ); + } else { + log::warn!( + "VISION: TLS 1.3 detected but cipher 0x{:04x} not supported for XTLS", + parsed.cipher_suite + ); } // Stop filtering early diff --git a/src/vless/vision_stream.rs b/src/vless/vision_stream.rs index 5b787ebd..cda99573 100644 --- a/src/vless/vision_stream.rs +++ b/src/vless/vision_stream.rs @@ -606,8 +606,8 @@ where assert!( self.inner_write_deframer.pending_bytes() == unprocessed_buf_len ); - // clear since the user will re-feed - self.inner_write_deframer.clear(); + // Discard the unaccepted tail, but keep TLS recognition for resubmission. + self.inner_write_deframer.discard_buffered(); return Poll::Ready(Ok(processed_len - existing_inner_len)); } Poll::Ready(Ok(())) => { @@ -857,6 +857,7 @@ impl AsyncStream for VisionStream { #[cfg(test)] mod tests { + mod decisions; mod native_errors; #[cfg(target_os = "linux")] mod splice; @@ -997,7 +998,7 @@ mod tests { impl AsyncStream for GatedIo {} - #[derive(Clone, Copy)] + #[derive(Clone, Copy, Debug)] enum Backend { Tls12, Tls13, diff --git a/src/vless/vision_stream/tests/decisions.rs b/src/vless/vision_stream/tests/decisions.rs new file mode 100644 index 00000000..254c9fd7 --- /dev/null +++ b/src/vless/vision_stream/tests/decisions.rs @@ -0,0 +1,630 @@ +use super::*; +use crate::vless::vision_pad::pad_with_uuid_and_command; +use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; +use tokio::time::{Duration, timeout}; + +type LegacyVision = VisionStream>; + +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"; + +fn record(content_type: u8, body: &[u8]) -> Vec { + let mut bytes = vec![content_type, 3, 3]; + bytes.extend_from_slice(&u16::try_from(body.len()).unwrap().to_be_bytes()); + bytes.extend_from_slice(body); + bytes +} + +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 { + let mut body = vec![3, 3]; + body.extend_from_slice(&[42; 32]); + body.push(session_id_len); + body.extend(std::iter::repeat_n(7, usize::from(session_id_len))); + body.extend_from_slice(&cipher.to_be_bytes()); + body.push(0); + body.extend_from_slice(&u16::try_from(extensions.len()).unwrap().to_be_bytes()); + body.extend_from_slice(extensions); + let mut handshake = vec![2]; + handshake.extend_from_slice(&u32::try_from(body.len()).unwrap().to_be_bytes()[1..]); + handshake.extend_from_slice(&body); + record(0x16, &handshake) +} + +fn decrypt_output(peer: &mut CryptoConnection, ciphertext: &[u8]) -> Vec { + use std::io::Read; + + feed_and_process_crypto_connection(peer, ciphertext).unwrap(); + let mut plaintext = Vec::new(); + if let Err(error) = peer.reader().read_to_end(&mut plaintext) { + assert_eq!(error.kind(), io::ErrorKind::WouldBlock); + } + plaintext +} + +async fn pair(backend: Backend) -> (LegacyVision, LegacyVision) { + // Independent matrix cases must not inherit an exhausted cooperative budget. + tokio::task::yield_now().await; + let (client, server) = connection_pair(backend); + let (left, right) = tokio::io::duplex(65536); + ( + VisionStream::new_client( + CryptoTlsStream::new(left, client, Some(TlsDeframer::new())), + [7; 16], + ) + .unwrap(), + VisionStream::new_server( + CryptoTlsStream::new(right, server, Some(TlsDeframer::new())), + [7; 16], + b"", + ) + .unwrap(), + ) +} + +async fn transfer(sender: &mut LegacyVision, receiver: &mut LegacyVision, data: &[u8]) { + timeout(Duration::from_secs(2), async { + sender.write_all(data).await.unwrap(); + sender.flush().await.unwrap(); + let mut received = vec![0; data.len()]; + receiver.read_exact(&mut received).await.unwrap(); + assert_eq!(received, data); + }) + .await + .expect("Vision transfer stalled"); +} + +fn assert_padding(client: &LegacyVision, server: &LegacyVision) { + for stream in [client, server] { + assert_eq!(stream.read_mode, VisionMode::PaddingTls); + assert_eq!(stream.write_mode, VisionMode::PaddingTls); + assert!(!stream.pending_tls_mode_switch); + } +} + +fn assert_terminal_write(stream: &VisionStream, direct: bool) { + if direct { + assert_eq!(stream.write_mode, VisionMode::Direct); + assert!(!stream.pending_tls_mode_switch); + } else { + // END stays pending until the next write, even after a flush. + assert_eq!(stream.write_mode, VisionMode::PaddingTls); + assert!(stream.pending_tls_mode_switch); + } +} + +async fn assert_decision( + client: &mut LegacyVision, + server: &mut LegacyVision, + direct: bool, + server_first: bool, +) { + assert_eq!( + client.filter.supports_xtls(), + direct, + "client: {:?}", + client.filter + ); + assert_eq!( + server.filter.supports_xtls(), + direct, + "server: {:?}", + server.filter + ); + let expected = if direct { + VisionMode::Direct + } else { + VisionMode::Tls + }; + let (first, second) = if server_first { + (server, client) + } else { + (client, server) + }; + transfer(first, second, APPLICATION).await; + assert_terminal_write(first, direct); + assert_eq!(second.read_mode, expected); + assert_eq!(first.read_mode, VisionMode::PaddingTls); + assert_eq!(second.write_mode, VisionMode::PaddingTls); + transfer(second, first, APPLICATION).await; + assert_terminal_write(second, direct); + for stream in [&*first, &*second] { + assert_eq!(stream.read_mode, expected); + } + // The receiver must follow the on-wire command, not merely the sender's local state. + transfer(first, second, b"payload after padding stops").await; + transfer(second, first, b"reverse payload after padding stops").await; + assert_eq!(first.write_mode, expected); + assert_eq!(second.write_mode, expected); +} + +#[tokio::test] +async fn legacy_direct_requires_tls13_and_a_supported_cipher() { + let cases = [ + (0x1301, TLS13_VERSION, true), + (0x1302, TLS13_VERSION, true), + (0x1303, TLS13_VERSION, true), + (0x1304, TLS13_VERSION, true), + (0x1305, TLS13_VERSION, false), + (0xc02f, TLS13_VERSION, false), + (0x1306, TLS13_VERSION, false), + (0x0a0a, TLS13_VERSION, false), + (0xc02f, b"".as_slice(), false), + (0x1301, b"".as_slice(), false), + (0x1301, b"\x00\x2b\x00\x02\x03\x03".as_slice(), false), + ]; + for backend in BACKENDS { + for (cipher, extensions, direct) in cases { + for server_first in [false, true] { + eprintln!( + "{backend:?}, cipher={cipher:#06x}, extensions={extensions:?}, server_first={server_first}" + ); + let (mut client, mut server) = pair(backend).await; + transfer(&mut client, &mut server, &client_hello()).await; + assert!(!client.filter.supports_xtls()); + assert!(!server.filter.supports_xtls()); + transfer( + &mut server, + &mut client, + &server_hello(cipher, 32, extensions), + ) + .await; + assert_eq!(client.filter.supports_xtls(), direct); + assert_eq!(server.filter.supports_xtls(), direct); + // ServerHello and compatibility CCS alone must not end padding. + transfer(&mut server, &mut client, CHANGE_CIPHER_SPEC).await; + assert_padding(&client, &server); + assert_decision(&mut client, &mut server, direct, server_first).await; + } + } + } +} + +#[tokio::test] +async fn legacy_direct_parses_session_ids_and_extension_boundaries() { + let unknown = b"\xfa\xfa\x00\x07\x00\x2b\x00\x02\x03\x04\x00"; + for backend in BACKENDS { + for session_id_len in [0, 1, 32] { + for (extensions, direct) in [ + (TLS13_VERSION.to_vec(), true), + ([unknown.as_slice(), TLS13_VERSION].concat(), true), + ([TLS13_VERSION, unknown.as_slice()].concat(), true), + (unknown.to_vec(), false), + ] { + eprintln!( + "{backend:?}, session_id_len={session_id_len}, extensions={extensions:?}" + ); + let (mut client, mut server) = pair(backend).await; + transfer(&mut client, &mut server, &client_hello()).await; + transfer( + &mut server, + &mut client, + &server_hello(0x1301, session_id_len, &extensions), + ) + .await; + assert_padding(&client, &server); + assert_decision(&mut client, &mut server, direct, false).await; + } + } + } +} + +#[tokio::test] +async fn legacy_malformed_server_hello_never_enables_direct() { + let valid = server_hello(0x1301, 32, TLS13_VERSION); + let mut long_handshake = valid.clone(); + long_handshake[6..9].copy_from_slice(&[0, 1, 0]); + let mut short_handshake = valid.clone(); + short_handshake[6..9].copy_from_slice(&[0, 0, 2]); + let mut long_extensions = valid.clone(); + long_extensions[79..81].copy_from_slice(&100u16.to_be_bytes()); + let mut short_extensions = valid.clone(); + short_extensions[79..81].copy_from_slice(&0u16.to_be_bytes()); + let mut retry = valid.clone(); + retry[11..43].copy_from_slice(&[ + 0xcf, 0x21, 0xad, 0x74, 0xe5, 0x9a, 0x61, 0x11, 0xbe, 0x1d, 0x8c, 0x02, 0x1e, 0x65, 0xb8, + 0x91, 0xc2, 0xa2, 0x11, 0x16, 0x7a, 0xbb, 0x8c, 0x5e, 0x07, 0x9e, 0x09, 0xe2, 0xc8, 0xa8, + 0x33, 0x9c, + ]); + let cases = [ + ("short hello", record(0x16, &[2, 0, 0, 2, 3, 3])), + ("handshake length", long_handshake), + ("undersized handshake length", short_handshake), + ("session ID length", server_hello(0x1301, 33, TLS13_VERSION)), + ("extensions length", long_extensions), + ("trailing data after extensions", short_extensions), + ( + "truncated extension header", + server_hello(0x1301, 32, &[0, 43, 0]), + ), + ( + "short selected version", + server_hello(0x1301, 32, &[0, 43, 0, 1, 3]), + ), + ( + "truncated selected version", + server_hello(0x1301, 32, &[0, 43, 0, 2, 3]), + ), + ("HelloRetryRequest", retry), + ]; + for backend in BACKENDS { + for (name, hello) in &cases { + eprintln!("{backend:?}, {name}"); + assert!( + crate::shadow_tls::parse_server_hello(hello).is_err(), + "{name}" + ); + let (mut client, mut server) = pair(backend).await; + transfer(&mut client, &mut server, &client_hello()).await; + transfer(&mut server, &mut client, hello).await; + assert!(!client.filter.supports_xtls()); + assert!(!server.filter.supports_xtls()); + assert_padding(&client, &server); + assert_decision(&mut client, &mut server, false, false).await; + } + } +} + +#[tokio::test] +async fn legacy_server_hello_write_splits_preserve_the_decision() { + for backend in BACKENDS { + for (cipher, direct) in [(0x1301, true), (0x1305, false)] { + let hello = server_hello(cipher, 32, TLS13_VERSION); + for split in 1..hello.len() { + eprintln!("{backend:?}, cipher={cipher:#06x}, split={split}"); + let (mut client, mut server) = pair(backend).await; + transfer(&mut client, &mut server, &client_hello()).await; + server.write_all(&hello[..split]).await.unwrap(); + server.flush().await.unwrap(); + assert!(!server.filter.supports_xtls()); + let mut byte = [0]; + let mut buf = ReadBuf::new(&mut byte); + assert!( + Pin::new(&mut client) + .poll_read(&mut Context::from_waker(noop_waker_ref()), &mut buf) + .is_pending() + ); + assert!(buf.filled().is_empty()); + assert_padding(&client, &server); + server.write_all(&hello[split..]).await.unwrap(); + server.flush().await.unwrap(); + let mut received = vec![0; hello.len()]; + timeout(Duration::from_secs(2), client.read_exact(&mut received)) + .await + .unwrap() + .unwrap(); + assert_eq!(received, hello); + assert_padding(&client, &server); + assert_decision(&mut client, &mut server, direct, false).await; + } + } + } +} + +#[tokio::test] +async fn legacy_server_hello_across_incoming_vision_frames_preserves_the_decision() { + for backend in BACKENDS { + for (cipher, direct) in [(0x1301, true), (0x1305, false)] { + let hello = server_hello(cipher, 32, TLS13_VERSION); + for split in 1..hello.len() { + eprintln!("incoming {backend:?}, cipher={cipher:#06x}, split={split}"); + tokio::task::yield_now().await; + let (session, mut peer) = connection_pair(backend); + let (io, mut wire) = tokio::io::duplex(65536); + let mut client = VisionStream::new_client( + CryptoTlsStream::new(io, session, Some(TlsDeframer::new())), + [7; 16], + ) + .unwrap(); + let request = client_hello(); + client.write_all(&request).await.unwrap(); + client.flush().await.unwrap(); + for (index, fragment) in [&hello[..split], &hello[split..]].into_iter().enumerate() + { + let mut padded = Vec::new(); + if index == 0 { + padded.extend_from_slice(&[0, 0]); + padded.extend_from_slice(&[7; 16]); + } + padded.push(COMMAND_CONTINUE); + padded.extend_from_slice(&u16::try_from(fragment.len()).unwrap().to_be_bytes()); + padded.extend_from_slice(&[0, 0]); + padded.extend_from_slice(fragment); + wire.write_all(&encrypt(&mut peer, &padded)).await.unwrap(); + let mut received = vec![0; fragment.len()]; + timeout(Duration::from_secs(2), client.read_exact(&mut received)) + .await + .unwrap() + .unwrap(); + assert_eq!(received, fragment); + assert_eq!(client.filter.supports_xtls(), index == 1 && direct); + assert_eq!(client.read_mode, VisionMode::PaddingTls); + assert_eq!(client.write_mode, VisionMode::PaddingTls); + } + client.write_all(APPLICATION).await.unwrap(); + client.flush().await.unwrap(); + assert_terminal_write(&client, direct); + + let mut ciphertext = Vec::new(); + loop { + let mut scratch = [0; 4096]; + let mut buf = ReadBuf::new(&mut scratch); + match Pin::new(&mut wire) + .poll_read(&mut Context::from_waker(noop_waker_ref()), &mut buf) + { + Poll::Ready(Ok(())) => { + assert!(!buf.filled().is_empty()); + ciphertext.extend_from_slice(buf.filled()); + } + Poll::Pending => break, + Poll::Ready(Err(error)) => panic!("{error}"), + } + } + let plaintext = decrypt_output(&mut peer, &ciphertext); + let decoded = VisionUnpadder::new([7; 16]).unpad(&plaintext).unwrap(); + let command = if direct { + UnpadCommand::Direct + } else { + UnpadCommand::End + }; + assert_eq!(decoded.command, Some(command)); + assert_eq!(decoded.content, [request.as_slice(), APPLICATION].concat()); + } + } + } +} + +#[tokio::test] +async fn legacy_incomplete_handshake_records_never_enable_direct() { + let hello = server_hello(0x1301, 32, TLS13_VERSION); + let handshake = &hello[5..]; + for backend in BACKENDS { + for split in [1, 4, 40, handshake.len() - 1] { + let (mut client, mut server) = pair(backend).await; + transfer(&mut client, &mut server, &client_hello()).await; + // Separate TLS records are not the same as splitting one record across writes. + let records = [ + record(0x16, &handshake[..split]), + record(0x16, &handshake[split..]), + ] + .concat(); + transfer(&mut server, &mut client, &records).await; + assert_padding(&client, &server); + assert_decision(&mut client, &mut server, false, false).await; + } + } +} + +#[tokio::test] +async fn legacy_application_splits_and_coalesced_tail_preserve_the_decision() { + for backend in BACKENDS { + for (cipher, direct) in [(0x1301, true), (0x1305, false)] { + for server_first in [false, true] { + for split in 1..APPLICATION.len() { + eprintln!( + "{backend:?}, cipher={cipher:#06x}, server_first={server_first}, split={split}" + ); + let (mut client, mut server) = pair(backend).await; + transfer(&mut client, &mut server, &client_hello()).await; + transfer( + &mut server, + &mut client, + &server_hello(cipher, 32, TLS13_VERSION), + ) + .await; + let expected = if direct { + VisionMode::Direct + } else { + VisionMode::Tls + }; + let (sender, receiver) = if server_first { + (&mut server, &mut client) + } else { + (&mut client, &mut server) + }; + sender.write_all(&APPLICATION[..split]).await.unwrap(); + sender.flush().await.unwrap(); + assert_eq!(sender.write_mode, VisionMode::PaddingTls); + assert!(!sender.pending_tls_mode_switch); + let tail = b"coalesced bytes after application record"; + let rest = [&APPLICATION[split..], tail].concat(); + sender.write_all(&rest).await.unwrap(); + sender.flush().await.unwrap(); + let mut received = vec![0; APPLICATION.len() + tail.len()]; + timeout(Duration::from_secs(2), receiver.read_exact(&mut received)) + .await + .unwrap() + .unwrap(); + assert_eq!(received, [APPLICATION, tail].concat()); + assert_eq!(sender.write_mode, expected); + assert_eq!(receiver.read_mode, expected); + transfer(receiver, sender, APPLICATION).await; + assert_terminal_write(receiver, direct); + assert_eq!(sender.read_mode, expected); + transfer(receiver, sender, b"reverse tail").await; + assert_eq!(receiver.write_mode, expected); + } + } + } + } +} + +#[tokio::test] +async fn legacy_filter_budget_does_not_enable_direct_after_end() { + let handshake = record(0x16, &[11, 0, 0, 0]); + for backend in BACKENDS { + for extra_records in [6, 7] { + let (mut client, mut server) = pair(backend).await; + transfer(&mut client, &mut server, &client_hello()).await; + for _ in 0..extra_records { + transfer(&mut server, &mut client, &handshake).await; + } + transfer( + &mut server, + &mut client, + &server_hello(0x1301, 32, TLS13_VERSION), + ) + .await; + if extra_records == 6 { + assert_decision(&mut client, &mut server, true, false).await; + } else { + assert_eq!(server.write_mode, VisionMode::Tls); + assert_eq!(client.read_mode, VisionMode::Tls); + assert!(!server.filter.supports_xtls()); + assert!(!client.filter.supports_xtls()); + transfer(&mut client, &mut server, APPLICATION).await; + assert_terminal_write(&client, false); + assert_eq!(server.read_mode, VisionMode::Tls); + transfer(&mut client, &mut server, b"still encrypted").await; + assert_eq!(client.write_mode, VisionMode::Tls); + transfer(&mut server, &mut client, APPLICATION).await; + assert_eq!(server.write_mode, VisionMode::Tls); + assert_eq!(client.read_mode, VisionMode::Tls); + } + } + } +} + +#[tokio::test] +async fn legacy_client_hello_alone_never_enables_direct() { + for backend in BACKENDS { + let (mut client, mut server) = pair(backend).await; + transfer(&mut client, &mut server, &client_hello()).await; + assert_padding(&client, &server); + transfer(&mut client, &mut server, APPLICATION).await; + assert_terminal_write(&client, false); + assert_eq!(server.read_mode, VisionMode::Tls); + transfer(&mut client, &mut server, b"still encrypted").await; + assert_eq!(client.write_mode, VisionMode::Tls); + } +} + +#[test] +fn legacy_pending_hello_write_preserves_decision_and_resubmitted_tail() { + use std::io::Read; + + for backend in BACKENDS { + for is_server in [false, true] { + for (cipher, direct) in [(0x1301, true), (0x1305, false)] { + let server_hello = server_hello(cipher, 32, TLS13_VERSION); + let hello = if is_server { + server_hello.clone() + } else { + client_hello() + }; + let tail = [APPLICATION, b"post-transition bytes"].concat(); + for retained in [0, 3, hello.len() - 1] { + for coalesced in [0, 3, tail.len()] { + eprintln!( + "{backend:?}, server={is_server}, cipher={cipher:#06x}, retained={retained}, coalesced={coalesced}" + ); + let (client, server) = connection_pair(backend); + let (session, mut peer) = if is_server { + (server, client) + } else { + (client, server) + }; + let reads = if is_server { + VecDeque::new() + } else { + let mut response = vec![0, 0]; + response.extend_from_slice(&pad_with_uuid_and_command( + &server_hello, + &[7; 16], + COMMAND_CONTINUE, + true, + )); + VecDeque::from([encrypt(&mut peer, &response).into()]) + }; + let gate = Arc::new(std::sync::Mutex::new(WriteGate::default())); + let io = GatedIo { + input: ScriptedIo { reads }, + gate: gate.clone(), + }; + let tls = CryptoTlsStream::new(io, session, Some(TlsDeframer::new())); + let mut stream = if is_server { + let mut stream = VisionStream::new_server(tls, [7; 16], b"").unwrap(); + // The gate targets the hello record, not the initial response header. + stream.vless_response_to_send = false; + stream + } else { + VisionStream::new_client(tls, [7; 16]).unwrap() + }; + let mut cx = Context::from_waker(noop_waker_ref()); + if retained != 0 { + assert!( + matches!(Pin::new(&mut stream).poll_write(&mut cx, &hello[..retained]), + Poll::Ready(Ok(n)) if n == retained) + ); + } + gate.lock().unwrap().blocked = true; + let input = [&hello[retained..], &tail[..coalesced]].concat(); + assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, &input), + Poll::Ready(Ok(n)) if n == hello.len() - retained)); + assert_eq!(stream.inner_write_deframer.pending_bytes(), 0); + assert!(gate.lock().unwrap().output.is_empty()); + assert!( + Pin::new(&mut stream) + .poll_write(&mut cx, &tail) + .is_pending() + ); + gate.lock().unwrap().blocked = false; + assert!(matches!( + Pin::new(&mut stream).poll_flush(&mut cx), + Poll::Ready(Ok(())) + )); + if !is_server { + assert_buffered_reads(&mut stream, &server_hello); + } + assert_eq!(stream.filter.supports_xtls(), direct); + assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, &tail), + Poll::Ready(Ok(n)) if n == APPLICATION.len())); + assert!(matches!( + Pin::new(&mut stream).poll_flush(&mut cx), + Poll::Ready(Ok(())) + )); + assert_terminal_write(&stream, direct); + + let ciphertext = std::mem::take(&mut gate.lock().unwrap().output); + let plaintext = decrypt_output(&mut peer, &ciphertext); + let decoded = VisionUnpadder::new([7; 16]).unpad(&plaintext).unwrap(); + assert_eq!(decoded.content, [hello.as_slice(), APPLICATION].concat()); + let command = if direct { + UnpadCommand::Direct + } else { + UnpadCommand::End + }; + assert_eq!(decoded.command, Some(command)); + + let raw_tail = &tail[APPLICATION.len()..]; + assert!( + matches!(Pin::new(&mut stream).poll_write(&mut cx, raw_tail), + Poll::Ready(Ok(n)) if n == raw_tail.len()) + ); + assert!(matches!( + Pin::new(&mut stream).poll_flush(&mut cx), + Poll::Ready(Ok(())) + )); + let output = std::mem::take(&mut gate.lock().unwrap().output); + if direct { + assert_eq!(output, raw_tail); + } else { + assert_eq!(stream.write_mode, VisionMode::Tls); + feed_and_process_crypto_connection(&mut peer, &output).unwrap(); + let mut received = vec![0; raw_tail.len()]; + peer.reader().read_exact(&mut received).unwrap(); + assert_eq!(received, raw_tail); + } + } + } + } + } + } +}