diff --git a/docs/Config_params/CONFIG_PARAMS.en.md b/docs/Config_params/CONFIG_PARAMS.en.md index 799bc3e..7b7cf4d 100644 --- a/docs/Config_params/CONFIG_PARAMS.en.md +++ b/docs/Config_params/CONFIG_PARAMS.en.md @@ -1971,7 +1971,7 @@ This document lists all configuration keys accepted by `config.toml`. ``` ## client_mss - **Constraints / validation**: `String`. Empty or omitted means do not change kernel MSS. Presets: `"extreme-low"` = `88`, `"tspu"` = `92`, `"2in8"` = `256`. Custom decimal strings must be within `88..=4096`. - - **Description**: Client-facing TCP MSS applied to TCP listener sockets before `listen(2)`, so Linux can announce it in SYN/ACK. This affects only proxy client TCP listeners, not API, metrics, Unix sockets, Telegram upstreams, ME sockets, or mask backend connections. Changes require listener restart/rebind. + - **Description**: Controls the segment size used for client connections. By default, this value is applied to the TCP listener and remains active for the whole connection. When `client_mss_bulk` is also set on Linux, `client_mss` is used only as the maximum chunk size for the initial authenticated FakeTLS response (`ServerHello`). This setting does not affect API, metrics, Unix sockets, Telegram upstreams, ME sockets, or mask backend connections. Changes require a listener restart/rebind. - **Operator note**: The two-tier `synlimit` profile does not require Telemt to disable MSS automatically. Operators that follow external host-tuning recipes should decide explicitly whether to leave MSS shaping enabled for handshake fragmentation or disable it for higher media throughput. - **Performance note**: Low MSS increases packet count predictably. Approximate segment multiplier is `ceil(1460 / client_mss)`. - **Example**: @@ -1982,7 +1982,7 @@ This document lists all configuration keys accepted by `config.toml`. ``` ## client_mss_bulk - **Constraints / validation**: `String`. Same grammar as [`client_mss`](#client_mss) (empty/omitted, presets `"extreme-low"`/`"tspu"`/`"2in8"`, or a decimal in `88..=4096`). - - **Description**: Optional bulk-phase MSS. When set, the low `client_mss` is applied only while the TLS handshake (including the DPI-inspected ServerHello) is sent; once the connection transitions to relaying, the client socket MSS is raised to `client_mss_bulk` for the bulk data phase. This keeps the anti-DPI handshake fragmentation but restores normal-size packets for payload, cutting outgoing packets-per-second by roughly the `client_mss` segment multiplier (e.g. ~10x with `"tspu"`). Useful on hosts whose abuse detection counts packets-per-second rather than bandwidth. When empty/omitted, the handshake MSS is kept for the whole connection (previous behavior). Linux only; a no-op elsewhere. + - **Description**: Enables separated MSS handling for `ServerHello` and for all other packets. The listener uses `client_mss_bulk` from the start of the connection, including when the client sends `ClientHello`. Telemt sends the initial authenticated FakeTLS response (`ServerHello`) in chunks no larger than `client_mss`. Normal MTProto traffic then continues with `client_mss_bulk`, without changing MSS on an established connection. This keeps the smaller `ServerHello` chunks while avoiding small segments for all later traffic. When this option is empty or omitted, `client_mss` applies to the whole connection. **Linux-only**. - **Example**: ```toml diff --git a/docs/Config_params/CONFIG_PARAMS.ru.md b/docs/Config_params/CONFIG_PARAMS.ru.md index ddcb881..7eaedf5 100644 --- a/docs/Config_params/CONFIG_PARAMS.ru.md +++ b/docs/Config_params/CONFIG_PARAMS.ru.md @@ -1893,7 +1893,7 @@ ``` ## client_mss - **Ограничения / валидация**: `String`. Пустое значение или отсутствие параметра означает, что Telemt не изменяет MSS, выбранный ядром. Поддерживаемые presets: `"extreme-low"` = `88`, `"tspu"` = `92`, `"2in8"` = `256`. Пользовательское десятичное значение должно быть строкой в диапазоне `88..=4096`. - - **Описание**: MSS для входящих TCP-соединений клиентов. Значение применяется к TCP listener-сокетам до `listen(2)`, чтобы Linux мог объявить его в SYN/ACK. Параметр влияет только на proxy client TCP listeners и не применяется к API, metrics, Unix sockets, Telegram upstreams, ME sockets или mask backend connections. Изменение требует restart/rebind listener’ов. + - **Описание**: Управляет размером сегментов в клиентских соединениях. По умолчанию значение применяется к TCP listener и действует в течение всего соединения. Если также задан `client_mss_bulk`, параметр `client_mss` используется только как максимальный размер фрагмента начального аутентифицированного FakeTLS-ответа (`ServerHello`). Настройка не влияет на API, metrics, Unix sockets, Telegram upstreams, ME sockets и mask backend connections. Изменение требует перезапуска или повторного создания listener. - **Operator note**: Two-tier `synlimit` profile больше не требует автоматического отключения MSS внутри Telemt. Оператор должен сам решить, оставлять MSS shaping для handshake fragmentation или отключать его ради более высокой скорости media. - **Performance note**: Низкий MSS предсказуемо увеличивает количество TCP-сегментов. Приблизительный multiplier: `ceil(1460 / client_mss)`. - **Пример**: @@ -1904,7 +1904,7 @@ ``` ## client_mss_bulk - **Ограничения / валидация**: `String`. Грамматика та же, что у [`client_mss`](#client_mss) (пусто/не задано, пресеты `"extreme-low"`/`"tspu"`/`"2in8"` либо десятичное число в диапазоне `88..=4096`). - - **Описание**: Необязательный MSS для bulk-фазы. Если задан, низкий `client_mss` применяется только на время TLS-handshake (включая инспектируемый DPI ServerHello); как только соединение переходит в фазу relay, MSS клиентского сокета поднимается до `client_mss_bulk` для передачи полезной нагрузки. Так сохраняется anti-DPI фрагментация handshake, но для данных возвращаются пакеты нормального размера — это снижает исходящий packets-per-second примерно во столько раз, каков segment multiplier у `client_mss` (например, ~10x для `"tspu"`). Полезно на хостингах, где abuse-детекция считает packets-per-second, а не полосу. Если пусто/не задано — MSS handshake сохраняется на всё соединение (прежнее поведение). Только Linux; на прочих платформах — no-op. + - **Описание**: Включает раздельную работу MSS для `ServerHello` и остального трафика. Listener использует `client_mss_bulk` с самого начала соединения, в том числе при получении `ClientHello`. Начальный аутентифицированный FakeTLS-ответ (`ServerHello`) Telemt отправляет фрагментами не больше `client_mss`. После этого обычный MTProto-трафик продолжает работать с `client_mss_bulk`, без изменения MSS уже установленного соединения. Так `ServerHello` остаётся разбит на небольшие части, а весь последующий трафик не ограничивается маленькими сегментами. Если параметр пуст или не задан, `client_mss` действует в течение всего соединения. **Работает только в Linux.** - **Пример**: ```toml diff --git a/src/config/types.rs b/src/config/types.rs index 4607732..970fde2 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -1619,12 +1619,11 @@ pub struct ServerConfig { #[serde(default)] pub client_mss: Option, - /// Client-facing TCP MSS to switch to AFTER the TLS handshake (ServerHello) - /// is sent. Lets `client_mss` fragment ONLY the handshake (the DPI-inspected - /// part) while the bulk transfer uses normal-size packets — avoids the ~10x - /// packets-per-second blowup that triggers anti-DDoS abuse blocks on - /// pps-policing hosts. Empty/omitted = keep the handshake MSS for the whole - /// connection (previous behavior). Same preset/int grammar as `client_mss`. + /// Client-facing TCP MSS for bulk traffic when initial-response fragmentation + /// is enabled on Linux. The listener uses this MSS from connection setup; + /// `client_mss` becomes the fragment size for the authenticated FakeTLS + /// response. Empty/omitted keeps `client_mss` connection-wide. Uses the same + /// preset/integer grammar as `client_mss`. #[serde(default)] pub client_mss_bulk: Option, @@ -2484,7 +2483,7 @@ impl ServerConfig { parse_client_mss(self.client_mss.as_deref()) } - /// Resolves the post-handshake (bulk transfer) client MSS, if configured. + /// Resolves the bulk-transfer client MSS, if configured. pub fn client_mss_bulk_value(&self) -> std::result::Result, String> { parse_client_mss(self.client_mss_bulk.as_deref()) } diff --git a/src/maestro/listeners.rs b/src/maestro/listeners.rs index 15cd31f..1a21779 100644 --- a/src/maestro/listeners.rs +++ b/src/maestro/listeners.rs @@ -30,10 +30,17 @@ use super::helpers::{ }; pub(crate) struct BoundListeners { - pub(crate) listeners: Vec<(TcpListener, bool)>, + pub(crate) listeners: Vec, pub(crate) has_unix_listener: bool, } +/// A TCP listener and the connection settings fixed when it was bound. +pub(crate) struct BoundTcpListener { + listener: TcpListener, + proxy_protocol: bool, + tls_response_fragment_size: Option, +} + fn listener_port_or_legacy(listener: &crate::config::ListenerConfig, config: &ProxyConfig) -> u16 { listener.port.unwrap_or(config.server.port) } @@ -51,6 +58,18 @@ fn mss_segment_multiplier(client_mss: u16) -> u16 { 1460u16.div_ceil(client_mss) } +#[cfg(any(target_os = "linux", test))] +fn tcp_mss_runtime_profile( + handshake_mss: Option, + bulk_mss: Option, +) -> (Option, Option) { + match (handshake_mss, bulk_mss) { + (Some(fragment_size), Some(listener_mss)) => (Some(listener_mss), Some(fragment_size)), + (listener_mss, None) => (listener_mss, None), + (None, Some(_)) => (None, None), + } +} + #[allow(clippy::too_many_arguments)] pub(crate) async fn bind_listeners( config: &Arc, @@ -82,6 +101,16 @@ pub(crate) async fn bind_listeners( ) .await; let mut listeners = Vec::new(); + let bulk_client_mss = match config.server.client_mss_bulk_value() { + Ok(value) => value, + Err(error) => { + warn!( + error = %error, + "Invalid bulk client MSS after config validation; disabling bulk MSS" + ); + None + } + }; for listener_conf in &config.server.listeners { let listener_port = listener_port_or_legacy(listener_conf, config); @@ -94,7 +123,7 @@ pub(crate) async fn bind_listeners( warn!(%addr, "Skipping IPv6 listener: IPv6 disabled by [network]"); continue; } - let client_mss = match listener_conf.effective_client_mss(&config.server) { + let configured_client_mss = match listener_conf.effective_client_mss(&config.server) { Ok(value) => value, Err(error) => { warn!( @@ -105,6 +134,11 @@ pub(crate) async fn bind_listeners( None } }; + #[cfg(target_os = "linux")] + let (client_mss, tls_response_fragment_size) = + tcp_mss_runtime_profile(configured_client_mss, bulk_client_mss); + #[cfg(not(target_os = "linux"))] + let (client_mss, tls_response_fragment_size) = (configured_client_mss, None); let options = ListenOptions { reuse_port: listener_conf.reuse_allow, ipv6_only: listener_conf.ip.is_ipv6(), @@ -125,6 +159,15 @@ pub(crate) async fn bind_listeners( "Client-facing TCP MSS configured" ); } + if let Some(fragment_size) = tls_response_fragment_size { + info!( + %addr, + fragment_size, + bulk_mss = client_mss, + segment_multiplier = mss_segment_multiplier(fragment_size), + "Initial FakeTLS response fragmentation configured" + ); + } let listener_proxy_protocol = listener_conf .proxy_protocol .unwrap_or(config.server.proxy_protocol); @@ -152,7 +195,11 @@ pub(crate) async fn bind_listeners( print_proxy_links(&public_host, link_port, config); } - listeners.push((listener, listener_proxy_protocol)); + listeners.push(BoundTcpListener { + listener, + proxy_protocol: listener_proxy_protocol, + tls_response_fragment_size, + }); } Err(e) => { if e.kind() == std::io::ErrorKind::AddrInUse { @@ -387,7 +434,7 @@ pub(crate) async fn bind_listeners( #[allow(clippy::too_many_arguments)] pub(crate) fn spawn_tcp_accept_loops( - listeners: Vec<(TcpListener, bool)>, + listeners: Vec, config_rx: watch::Receiver>, admission_rx: watch::Receiver, stats: Arc, @@ -404,7 +451,10 @@ pub(crate) fn spawn_tcp_accept_loops( shared: Arc, max_connections: Arc, ) { - for (listener, listener_proxy_protocol) in listeners { + for bound_listener in listeners { + let listener = bound_listener.listener; + let listener_proxy_protocol = bound_listener.proxy_protocol; + let tls_response_fragment_size = bound_listener.tls_response_fragment_size; let mut config_rx: watch::Receiver> = config_rx.clone(); let admission_rx_tcp = admission_rx.clone(); let stats = stats.clone(); @@ -513,6 +563,7 @@ pub(crate) fn spawn_tcp_accept_loops( #[cfg(unix)] raw_fd, rst_mode, + tls_response_fragment_size, ) .run() .await @@ -603,3 +654,26 @@ pub(crate) fn spawn_tcp_accept_loops( }); } } + +#[cfg(test)] +mod tests { + use super::tcp_mss_runtime_profile; + + #[test] + fn client_mss_without_bulk_remains_connection_wide() { + assert_eq!(tcp_mss_runtime_profile(Some(92), None), (Some(92), None)); + } + + #[test] + fn client_mss_with_bulk_uses_bulk_listener_and_fragments_initial_response() { + assert_eq!( + tcp_mss_runtime_profile(Some(92), Some(1400)), + (Some(1400), Some(92)) + ); + } + + #[test] + fn bulk_mss_without_handshake_mss_does_not_enable_shaping() { + assert_eq!(tcp_mss_runtime_profile(None, Some(1400)), (None, None)); + } +} diff --git a/src/proxy/client.rs b/src/proxy/client.rs index bd30e5b..f4c5367 100644 --- a/src/proxy/client.rs +++ b/src/proxy/client.rs @@ -109,7 +109,8 @@ use crate::transport::{UpstreamManager, configure_client_socket, parse_proxy_pro use crate::proxy::direct_relay::handle_via_direct_with_shared; use crate::proxy::handshake::{ - HandshakeSuccess, handle_mtproto_handshake_with_shared, handle_tls_handshake_with_shared, + HandshakeSuccess, TlsResponseWriteOptions, handle_mtproto_handshake_with_shared, + handle_tls_handshake_with_shared, handle_tls_handshake_with_shared_and_options, }; #[cfg(test)] use crate::proxy::handshake::{handle_mtproto_handshake, handle_tls_handshake}; @@ -989,6 +990,7 @@ pub struct RunningClientHandler { #[cfg(unix)] raw_fd: std::os::unix::io::RawFd, rst_on_close: crate::config::RstOnCloseMode, + tls_response_fragment_size: Option, } impl ClientHandler { @@ -1036,6 +1038,7 @@ impl ClientHandler { #[cfg(unix)] raw_fd, crate::config::RstOnCloseMode::Off, + None, ) } @@ -1060,6 +1063,7 @@ impl ClientHandler { real_peer_report: Arc>>, #[cfg(unix)] raw_fd: std::os::unix::io::RawFd, rst_on_close: crate::config::RstOnCloseMode, + tls_response_fragment_size: Option, ) -> RunningClientHandler { let normalized_peer = normalize_ip(peer); RunningClientHandler { @@ -1084,6 +1088,7 @@ impl ClientHandler { #[cfg(unix)] raw_fd, rst_on_close, + tls_response_fragment_size, } } } @@ -1105,12 +1110,6 @@ impl RunningClientHandler { #[cfg(unix)] let raw_fd = self.raw_fd; let rst_on_close = self.rst_on_close; - // MSS for the bulk data phase: once the handshake (incl. ServerHello) is - // sent, restore a normal MSS so only the handshake stays fragmented by the - // low listener `client_mss`. Cuts pps ~10x (anti-DDoS abuse on pps-policing - // hosts like FastVPS). None = keep handshake MSS for the whole connection. - #[cfg(unix)] - let bulk_mss: Option = self.config.server.client_mss_bulk_value().ok().flatten(); let outcome = match self.do_handshake().await? { Some(outcome) => outcome, @@ -1124,14 +1123,6 @@ impl RunningClientHandler { if matches!(rst_on_close, crate::config::RstOnCloseMode::Errors) { let _ = crate::transport::socket::clear_linger_fd(raw_fd); } - // Handshake (ServerHello) done — raise MSS for bulk transfer. - #[cfg(unix)] - if let Some(mss) = bulk_mss { - if let Err(e) = crate::transport::socket::set_tcp_mss_fd(raw_fd, u32::from(mss)) - { - debug!(error = %e, "Failed to raise bulk MSS; keeping handshake MSS"); - } - } fut.await } HandshakeOutcome::NeedsMasking(fut) => fut.await, @@ -1412,50 +1403,58 @@ impl RunningClientHandler { let (read_half, write_half) = self.stream.into_split(); - let (mut tls_reader, tls_writer, tls_user) = match handle_tls_handshake_with_shared( - &handshake, - read_half, - write_half, - peer, - &config, - &replay_checker, - &self.rng, - self.tls_cache.clone(), - self.shared.as_ref(), - ) - .await - { - HandshakeResult::Success(result) => result, - HandshakeResult::BadClient { reader, writer } => { - stats.increment_connects_bad_with_class("tls_handshake_bad_client"); - record_tls_fingerprint_bad_or_probe( - stats.as_ref(), - &config, - peer.ip(), - tls_fingerprint.as_ref(), - ); - return Ok(masking_outcome( - reader, - writer, - handshake.clone(), - peer, - local_addr, - config.clone(), - self.beobachten.clone(), - self.shared.clone(), - )); - } - HandshakeResult::Error(e) => { - record_tls_fingerprint_bad_or_probe( - stats.as_ref(), - &config, - peer.ip(), - tls_fingerprint.as_ref(), - ); - increment_bad_on_unknown_tls_sni(stats.as_ref(), &e); - return Err(e); - } - }; + #[cfg(target_os = "linux")] + let response_write_options = + TlsResponseWriteOptions::tcp(self.raw_fd, self.tls_response_fragment_size); + #[cfg(not(target_os = "linux"))] + let response_write_options = TlsResponseWriteOptions::default(); + + let (mut tls_reader, tls_writer, tls_user) = + match handle_tls_handshake_with_shared_and_options( + &handshake, + read_half, + write_half, + peer, + &config, + &replay_checker, + &self.rng, + self.tls_cache.clone(), + self.shared.as_ref(), + response_write_options, + ) + .await + { + HandshakeResult::Success(result) => result, + HandshakeResult::BadClient { reader, writer } => { + stats.increment_connects_bad_with_class("tls_handshake_bad_client"); + record_tls_fingerprint_bad_or_probe( + stats.as_ref(), + &config, + peer.ip(), + tls_fingerprint.as_ref(), + ); + return Ok(masking_outcome( + reader, + writer, + handshake.clone(), + peer, + local_addr, + config.clone(), + self.beobachten.clone(), + self.shared.clone(), + )); + } + HandshakeResult::Error(e) => { + record_tls_fingerprint_bad_or_probe( + stats.as_ref(), + &config, + peer.ip(), + tls_fingerprint.as_ref(), + ); + increment_bad_on_unknown_tls_sni(stats.as_ref(), &e); + return Err(e); + } + }; record_tls_fingerprint_auth_success( stats.as_ref(), &config, diff --git a/src/proxy/handshake.rs b/src/proxy/handshake.rs index f9f55de..134cb80 100644 --- a/src/proxy/handshake.rs +++ b/src/proxy/handshake.rs @@ -59,6 +59,26 @@ const OVERLOAD_CANDIDATE_BUDGET_UNHINTED: usize = 8; const EXPENSIVE_INVALID_SCAN_SATURATION_THRESHOLD: usize = 64; const RECENT_USER_RING_SCAN_LIMIT: usize = 32; +/// Controls how the authenticated FakeTLS response is written to a client. +#[derive(Clone, Copy, Debug, Default)] +pub(crate) struct TlsResponseWriteOptions { + #[cfg(target_os = "linux")] + socket_fd: Option, + #[cfg(target_os = "linux")] + fragment_size: Option, +} + +impl TlsResponseWriteOptions { + /// Creates Linux TCP response fragmentation options for an accepted socket. + #[cfg(target_os = "linux")] + pub(crate) fn tcp(fd: std::os::unix::io::RawFd, fragment_size: Option) -> Self { + Self { + socket_fd: fragment_size.map(|_| fd), + fragment_size, + } + } +} + #[cfg(test)] const AUTH_PROBE_BACKOFF_BASE_MS: u64 = 1; #[cfg(not(test))] @@ -925,6 +945,7 @@ where rng, tls_cache, shared.as_ref(), + TlsResponseWriteOptions::default(), ) .await } @@ -954,6 +975,39 @@ where rng, tls_cache, shared, + TlsResponseWriteOptions::default(), + ) + .await +} + +/// Handles FakeTLS with optional initial-response fragmentation on a TCP socket. +pub(crate) async fn handle_tls_handshake_with_shared_and_options( + handshake: &[u8], + reader: R, + writer: W, + peer: SocketAddr, + config: &ProxyConfig, + replay_checker: &ReplayChecker, + rng: &SecureRandom, + tls_cache: Option>, + shared: &ProxySharedState, + response_write_options: TlsResponseWriteOptions, +) -> HandshakeResult<(FakeTlsReader, FakeTlsWriter, String), R, W> +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + handle_tls_handshake_impl( + handshake, + reader, + writer, + peer, + config, + replay_checker, + rng, + tls_cache, + shared, + response_write_options, ) .await } @@ -968,6 +1022,7 @@ async fn handle_tls_handshake_impl( rng: &SecureRandom, tls_cache: Option>, shared: &ProxySharedState, + response_write_options: TlsResponseWriteOptions, ) -> HandshakeResult<(FakeTlsReader, FakeTlsWriter, String), R, W> where R: AsyncRead + Unpin, @@ -1468,16 +1523,11 @@ where debug!(peer = %peer, response_len = response.len(), "Sending TLS ServerHello"); - if let Err(e) = writer.write_all(&response).await { + if let Err(e) = write_tls_response(&mut writer, &response, response_write_options).await { warn!(peer = %peer, error = %e, "Failed to write TLS ServerHello"); return HandshakeResult::Error(ProxyError::Io(e)); } - if let Err(e) = writer.flush().await { - warn!(peer = %peer, error = %e, "Failed to flush TLS ServerHello"); - return HandshakeResult::Error(ProxyError::Io(e)); - } - debug!( peer = %peer, user = %validated_user, @@ -1498,6 +1548,26 @@ where )) } +async fn write_tls_response( + writer: &mut W, + response: &[u8], + options: TlsResponseWriteOptions, +) -> std::io::Result<()> { + #[cfg(target_os = "linux")] + if let (Some(fd), Some(fragment_size)) = (options.socket_fd, options.fragment_size) { + return crate::transport::socket::send_tcp_fragmented_fd( + fd, + response, + usize::from(fragment_size), + ) + .await; + } + + let _ = options; + writer.write_all(response).await?; + writer.flush().await +} + /// Handle MTProto obfuscation handshake #[cfg(test)] pub async fn handle_mtproto_handshake( diff --git a/src/proxy/tests/client_security_tests.rs b/src/proxy/tests/client_security_tests.rs index e00e5fc..338129d 100644 --- a/src/proxy/tests/client_security_tests.rs +++ b/src/proxy/tests/client_security_tests.rs @@ -1940,6 +1940,98 @@ async fn fragmented_tls_mtproto_with_interleaved_ccs_is_accepted() { assert_eq!(success.dc_idx, 2); } +#[cfg(target_os = "linux")] +#[tokio::test] +async fn chunked_tls_response_preserves_complete_records_and_configured_mss() { + use crate::proxy::handshake::{ + TlsResponseWriteOptions, handle_tls_handshake_with_shared_and_options, + }; + use crate::proxy::shared_state::ProxySharedState; + use crate::transport::socket::{ListenOptions, create_listener}; + use std::os::fd::{AsRawFd, BorrowedFd}; + + let options = ListenOptions { + reuse_port: false, + client_mss: Some(1400), + ..Default::default() + }; + let socket = create_listener("127.0.0.1:0".parse().unwrap(), &options).unwrap(); + let listener = TcpListener::from_std(socket.into()).unwrap(); + let listener_addr = listener.local_addr().unwrap(); + let mut client = TcpStream::connect(listener_addr).await.unwrap(); + let (server, peer) = listener.accept().await.unwrap(); + + let secret_hex = "56565656565656565656565656565656"; + let secret = [0x56u8; 16]; + let client_hello = make_valid_tls_client_hello(&secret, 0); + let mut config = ProxyConfig::default(); + config.general.beobachten = false; + config.access.ignore_time_skew = true; + config + .access + .users + .insert("user".to_string(), secret_hex.to_string()); + + let replay_checker = ReplayChecker::new(128, Duration::from_secs(60)); + let rng = SecureRandom::new(); + let shared = ProxySharedState::new(); + let raw_fd = server.as_raw_fd(); + let mss_before = socket2::SockRef::from(&server).tcp_mss().unwrap(); + let (read_half, write_half) = server.into_split(); + + let (tls_reader, tls_writer, tls_user) = match handle_tls_handshake_with_shared_and_options( + &client_hello, + read_half, + write_half, + peer, + &config, + &replay_checker, + &rng, + None, + &shared, + TlsResponseWriteOptions::tcp(raw_fd, Some(92)), + ) + .await + { + HandshakeResult::Success(result) => result, + _ => panic!("expected successful TLS handshake"), + }; + + // SAFETY: both split halves still own the accepted socket while it is borrowed. + let borrowed_fd = unsafe { BorrowedFd::borrow_raw(raw_fd) }; + let mss_after = socket2::SockRef::from(&borrowed_fd).tcp_mss().unwrap(); + assert_eq!( + mss_after, mss_before, + "chunked response writes must not change the configured socket MSS" + ); + assert_eq!(tls_user, "user"); + + drop(tls_reader); + let mut writer = tls_writer.into_inner(); + writer.shutdown().await.unwrap(); + + let mut response = Vec::new(); + client.read_to_end(&mut response).await.unwrap(); + assert!(!response.is_empty()); + + let mut offset = 0; + while offset < response.len() { + assert!(response.len() - offset >= 5, "truncated TLS record header"); + let payload_len = u16::from_be_bytes([response[offset + 3], response[offset + 4]]) as usize; + offset += 5; + assert!( + response.len() - offset >= payload_len, + "truncated TLS record payload" + ); + offset += payload_len; + } + assert_eq!( + offset, + response.len(), + "chunked response writes must preserve complete TLS records" + ); +} + #[tokio::test] async fn valid_tls_path_does_not_fall_back_to_mask_backend() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/src/transport/socket.rs b/src/transport/socket.rs index 2a6e762..839fcf9 100644 --- a/src/transport/socket.rs +++ b/src/transport/socket.rs @@ -125,36 +125,96 @@ pub fn clear_linger_fd(fd: std::os::unix::io::RawFd) -> Result<()> { Ok(()) } -/// Raise the TCP MSS on an already-accepted connection's fd. Used to fragment -/// ONLY the TLS handshake (via a low listener MSS) and then restore a normal MSS -/// for the bulk (post-handshake) data phase — cuts packets-per-second ~10x without losing the -/// DPI evasion that the fragmented ServerHello provides. No-op safe: errors are -/// returned to the caller, which logs and continues with the handshake MSS. #[cfg(target_os = "linux")] -pub fn set_tcp_mss_fd(fd: std::os::unix::io::RawFd, mss: u32) -> Result<()> { - use std::io::Error; - let mss = i32::try_from(mss) - .map_err(|_| Error::new(std::io::ErrorKind::InvalidInput, "bulk MSS out of range"))?; - // Direct setsockopt(TCP_MAXSEG) — same pattern as the TCP_USER_TIMEOUT call - // above; avoids socket2 method-name drift across versions. +fn force_tcp_push(fd: std::os::unix::io::RawFd) -> Result<()> { + let enabled: libc::c_int = 1; let rc = unsafe { libc::setsockopt( fd, libc::IPPROTO_TCP, - libc::TCP_MAXSEG, - &mss as *const libc::c_int as *const libc::c_void, + libc::TCP_NODELAY, + &enabled as *const libc::c_int as *const libc::c_void, std::mem::size_of::() as libc::socklen_t, ) }; if rc != 0 { - return Err(Error::last_os_error()); + return Err(std::io::Error::last_os_error()); } Ok(()) } -/// Non-Linux stub: MSS shaping only on Linux (TCP_MAXSEG). -#[cfg(all(unix, not(target_os = "linux")))] -pub fn set_tcp_mss_fd(_fd: std::os::unix::io::RawFd, _mss: u32) -> Result<()> { +/// Sends an initial TCP response in collapse-resistant segments on Linux. +/// +/// `fd` must refer to a connected, nonblocking TCP socket and remain valid for +/// the duration of this call. The caller retains ownership of the original fd. +/// Each successful `send(2)` is marked with `MSG_EOR`; Linux preserves that mark +/// when deciding whether adjacent SKBs may be collapsed, including retransmits. +/// Re-applying `TCP_NODELAY` forces each marked SKB out despite auto-corking. +#[cfg(target_os = "linux")] +pub(crate) async fn send_tcp_fragmented_fd( + fd: std::os::unix::io::RawFd, + data: &[u8], + fragment_size: usize, +) -> Result<()> { + use std::io::{Error, ErrorKind}; + use std::os::fd::{AsRawFd, BorrowedFd}; + use tokio::io::Interest; + use tokio::io::unix::AsyncFd; + + if fragment_size == 0 { + return Err(Error::new( + ErrorKind::InvalidInput, + "TCP fragment size must be greater than zero", + )); + } + if data.is_empty() { + return Ok(()); + } + if fd < 0 { + return Err(Error::from_raw_os_error(libc::EBADF)); + } + + // SAFETY: the caller guarantees that fd remains open for this call. + let borrowed_fd = unsafe { BorrowedFd::borrow_raw(fd) }; + let duplicated_fd = borrowed_fd.try_clone_to_owned()?; + let async_fd = AsyncFd::with_interest(duplicated_fd, Interest::WRITABLE)?; + + for fragment in data.chunks(fragment_size) { + let mut offset = 0; + while offset < fragment.len() { + let mut writable = async_fd.writable().await?; + let sent = match writable.try_io(|inner| { + let remaining = &fragment[offset..]; + let sent = unsafe { + libc::send( + inner.get_ref().as_raw_fd(), + remaining.as_ptr().cast::(), + remaining.len(), + libc::MSG_DONTWAIT | libc::MSG_EOR | libc::MSG_NOSIGNAL, + ) + }; + if sent < 0 { + Err(Error::last_os_error()) + } else if sent == 0 { + Err(Error::new( + ErrorKind::WriteZero, + "fragmented TCP send returned zero", + )) + } else { + Ok(sent as usize) + } + }) { + Ok(Ok(sent)) => sent, + Ok(Err(error)) if error.kind() == ErrorKind::Interrupted => continue, + Ok(Err(error)) => return Err(error), + Err(_) => continue, + }; + + offset += sent; + force_tcp_push(async_fd.get_ref().as_raw_fd())?; + } + } + Ok(()) } @@ -710,4 +770,72 @@ mod tests { }; assert_eq!(mss, 256); } + + #[cfg(target_os = "linux")] + #[tokio::test] + async fn test_chunked_send_preserves_stream_and_configured_mss() { + use std::os::fd::AsRawFd; + + let options = ListenOptions { + reuse_port: false, + client_mss: Some(1400), + ..Default::default() + }; + let socket = match create_listener("127.0.0.1:0".parse().unwrap(), &options) { + Ok(socket) => socket, + Err(e) if e.kind() == ErrorKind::PermissionDenied => return, + Err(e) => panic!("create_listener failed: {e}"), + }; + let listener = TcpListener::from_std(socket.into()).unwrap(); + let addr = listener.local_addr().unwrap(); + let client = TcpStream::connect(addr).await.unwrap(); + let (mut server, _) = listener.accept().await.unwrap(); + + let initial_response: Vec = (0..4096).map(|value| (value % 251) as u8).collect(); + let bulk_payload = vec![0xA5; 8192]; + let expected_len = initial_response.len() + bulk_payload.len(); + let reader = tokio::spawn(async move { + let mut client = client; + let mut received = vec![0; expected_len]; + client.read_exact(&mut received).await.unwrap(); + received + }); + + let mss_before = socket2::SockRef::from(&server).tcp_mss().unwrap(); + send_tcp_fragmented_fd(server.as_raw_fd(), &initial_response, 92) + .await + .unwrap(); + server.write_all(&bulk_payload).await.unwrap(); + let mss_after = socket2::SockRef::from(&server).tcp_mss().unwrap(); + + let mut expected = initial_response; + expected.extend_from_slice(&bulk_payload); + assert_eq!( + reader.await.unwrap(), + expected, + "chunked send must preserve the byte stream" + ); + assert_eq!( + mss_after, mss_before, + "chunked send must not change the configured socket MSS" + ); + } + + #[cfg(target_os = "linux")] + #[tokio::test] + async fn test_chunked_send_rejects_zero_fragment_size() { + let error = send_tcp_fragmented_fd(-1, b"response", 0) + .await + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidInput); + } + + #[cfg(target_os = "linux")] + #[tokio::test] + async fn test_chunked_send_rejects_invalid_fd() { + let error = send_tcp_fragmented_fd(-1, b"response", 92) + .await + .unwrap_err(); + assert_eq!(error.raw_os_error(), Some(libc::EBADF)); + } }