Restore single-record TLS-F primary application flight

Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>

src/config/hot_reload.rs

@@ -16,10 +16,12 @@
 //! | `general` | `telemetry` / `me_*_policy`    | Applied immediately                            |
 //! | `network` | `dns_overrides`                | Applied immediately                            |
 //! | `access`  | All user/quota fields          | Effective immediately                          |
+//! | `server.listeners` | `synlimit*` for existing endpoints | Netfilter rules reconciled immediately |
 //!
 //! Fields that require re-binding sockets (`server.listeners`, legacy
 //! `server.port`, `censorship.*`, `network.*`, `use_middle_proxy`) are **not**
-//! applied; a warning is emitted.
+//! applied, except for SYN limiter fields on unchanged listener endpoints; a
+//! warning is emitted.
 //! Non-hot changes are never mixed into the runtime config snapshot.
 
 use std::collections::BTreeSet;
@@ -34,7 +36,8 @@ use tracing::{error, info, warn};
 
 use super::load::{LoadedConfig, ProxyConfig};
 use crate::config::{
-    LogLevel, MeBindStaleMode, MeFloorMode, MeSocksKdfPolicy, MeTelemetryLevel, MeWriterPickMode,
+    ListenerConfig, LogLevel, MeBindStaleMode, MeFloorMode, MeSocksKdfPolicy, MeTelemetryLevel,
+    MeWriterPickMode, SynLimitMode,
 };
 
 const HOT_RELOAD_DEBOUNCE: Duration = Duration::from_millis(50);
@@ -131,6 +134,17 @@ pub struct HotFields {
     pub user_max_unique_ips_global_each: usize,
     pub user_max_unique_ips_mode: crate::config::UserMaxUniqueIpsMode,
     pub user_max_unique_ips_window_secs: u64,
+    pub listener_synlimit: Vec<ListenerSynLimitHotFields>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ListenerSynLimitHotFields {
+    pub ip: IpAddr,
+    pub port: Option<u16>,
+    pub synlimit: SynLimitMode,
+    pub synlimit_seconds: u32,
+    pub synlimit_hitcount: u32,
+    pub synlimit_burst: u32,
 }
 
 impl HotFields {
@@ -260,6 +274,25 @@ impl HotFields {
             user_max_unique_ips_global_each: cfg.access.user_max_unique_ips_global_each,
             user_max_unique_ips_mode: cfg.access.user_max_unique_ips_mode,
             user_max_unique_ips_window_secs: cfg.access.user_max_unique_ips_window_secs,
+            listener_synlimit: cfg
+                .server
+                .listeners
+                .iter()
+                .map(ListenerSynLimitHotFields::from_listener)
+                .collect(),
+        }
+    }
+}
+
+impl ListenerSynLimitHotFields {
+    fn from_listener(listener: &ListenerConfig) -> Self {
+        Self {
+            ip: listener.ip,
+            port: listener.port,
+            synlimit: listener.synlimit,
+            synlimit_seconds: listener.synlimit_seconds,
+            synlimit_hitcount: listener.synlimit_hitcount,
+            synlimit_burst: listener.synlimit_burst,
         }
     }
 }
@@ -566,6 +599,7 @@ fn overlay_hot_fields(old: &ProxyConfig, new: &ProxyConfig) -> ProxyConfig {
     cfg.access.user_max_unique_ips_global_each = new.access.user_max_unique_ips_global_each;
     cfg.access.user_max_unique_ips_mode = new.access.user_max_unique_ips_mode;
     cfg.access.user_max_unique_ips_window_secs = new.access.user_max_unique_ips_window_secs;
+    overlay_listener_synlimit_fields(&mut cfg.server.listeners, &new.server.listeners);
 
     if cfg.rebuild_runtime_user_auth().is_err() {
         cfg.runtime_user_auth = None;
@@ -574,6 +608,21 @@ fn overlay_hot_fields(old: &ProxyConfig, new: &ProxyConfig) -> ProxyConfig {
     cfg
 }
 
+fn overlay_listener_synlimit_fields(old: &mut [ListenerConfig], new: &[ListenerConfig]) {
+    if old.len() != new.len() {
+        return;
+    }
+    for (old_listener, new_listener) in old.iter_mut().zip(new.iter()) {
+        if old_listener.ip != new_listener.ip || old_listener.port != new_listener.port {
+            continue;
+        }
+        old_listener.synlimit = new_listener.synlimit;
+        old_listener.synlimit_seconds = new_listener.synlimit_seconds;
+        old_listener.synlimit_hitcount = new_listener.synlimit_hitcount;
+        old_listener.synlimit_burst = new_listener.synlimit_burst;
+    }
+}
+
 /// Warn if any non-hot fields changed (require restart).
 fn warn_non_hot_changes(old: &ProxyConfig, new: &ProxyConfig, non_hot_changed: bool) {
     let mut warned = false;
@@ -850,6 +899,13 @@ fn log_changes(
         );
     }
 
+    if old_hot.listener_synlimit != new_hot.listener_synlimit {
+        info!(
+            "config reload: server.listeners SYN limiter updated ({} listeners)",
+            new_hot.listener_synlimit.len()
+        );
+    }
+
     if old_hot.desync_all_full != new_hot.desync_all_full {
         info!(
             "config reload: desync_all_full: {} → {}",

src/synlimit_control.rs

@@ -79,14 +79,17 @@ pub(crate) fn spawn_synlimit_controller(config_rx: watch::Receiver<Arc<ProxyConf
     }
 
     tokio::spawn(async move {
-        wait_for_config_channel_close(config_rx).await;
+        wait_for_config_channel_close_and_reconcile(config_rx).await;
         clear_synlimit_rules_all_backends().await;
     });
 }
 
-async fn wait_for_config_channel_close(mut config_rx: watch::Receiver<Arc<ProxyConfig>>) {
+async fn wait_for_config_channel_close_and_reconcile(
+    mut config_rx: watch::Receiver<Arc<ProxyConfig>>,
+) {
     while config_rx.changed().await.is_ok() {
-        config_rx.borrow_and_update();
+        let cfg = config_rx.borrow_and_update().clone();
+        reconcile_synlimit_rules(&cfg).await;
     }
 }
 
@@ -183,10 +186,6 @@ async fn apply_iptables_synlimit_rules_for_binary(
     if targets.is_empty() {
         return Ok(());
     }
-    if !command_exists(binary) {
-        return Err(format!("{binary} is not available"));
-    }
-
     let _ = run_command(binary, &["-t", "filter", "-N", IPTABLES_CHAIN], None).await;
     run_command(binary, &["-t", "filter", "-F", IPTABLES_CHAIN], None).await?;
     if run_command(
@@ -316,9 +315,6 @@ fn synlimit_rate_arg(seconds: u32, hitcount: u32) -> String {
 }
 
 async fn clear_iptables_synlimit_rules_for_binary(binary: &str) {
-    if !command_exists(binary) {
-        return;
-    }
     for _ in 0..8 {
         if run_command(
             binary,
@@ -336,10 +332,6 @@ async fn clear_iptables_synlimit_rules_for_binary(binary: &str) {
 }
 
 async fn apply_nft_synlimit_rules(targets: &SynLimitTargets) -> Result<(), String> {
-    if !command_exists("nft") {
-        return Err("nft is not available".to_string());
-    }
-
     let families = detect_nft_table_families().await;
     for plan in nft_apply_plan(families, &targets.nft_v4, &targets.nft_v6) {
         let script = nft_synlimit_script(plan);
@@ -448,9 +440,6 @@ fn nft_synlimit_script(plan: NftApplyPlan<'_>) -> String {
 }
 
 async fn clear_nft_synlimit_rules_all_families() {
-    if !command_exists("nft") {
-        return;
-    }
     for family in [NftFamily::Inet, NftFamily::Ip, NftFamily::Ip6] {
         let _ = run_command(
             "nft",
@@ -462,10 +451,10 @@ async fn clear_nft_synlimit_rules_all_families() {
 }
 
 async fn run_command(binary: &str, args: &[&str], stdin: Option<String>) -> Result<(), String> {
-    if !command_exists(binary) {
+    let Some(command_path) = resolve_command(binary) else {
         return Err(format!("{binary} is not available"));
-    }
-    let mut command = Command::new(binary);
+    };
+    let mut command = Command::new(command_path);
     command.args(args);
     if stdin.is_some() {
         command.stdin(std::process::Stdio::piped());
@@ -499,10 +488,10 @@ async fn run_command(binary: &str, args: &[&str], stdin: Option<String>) -> Resu
 }
 
 async fn run_command_stdout(binary: &str, args: &[&str]) -> Result<String, String> {
-    if !command_exists(binary) {
+    let Some(command_path) = resolve_command(binary) else {
         return Err(format!("{binary} is not available"));
-    }
-    let output = Command::new(binary)
+    };
+    let output = Command::new(command_path)
         .args(args)
         .output()
         .await
@@ -518,14 +507,14 @@ async fn run_command_stdout(binary: &str, args: &[&str]) -> Result<String, Strin
     })
 }
 
-fn command_exists(binary: &str) -> bool {
-    let Some(path_var) = std::env::var_os("PATH") else {
-        return false;
-    };
-    std::env::split_paths(&path_var).any(|dir| {
-        let candidate: PathBuf = dir.join(binary);
-        candidate.exists() && candidate.is_file()
-    })
+fn resolve_command(binary: &str) -> Option<PathBuf> {
+    let mut dirs = std::env::var_os("PATH")
+        .map(|path| std::env::split_paths(&path).collect::<Vec<_>>())
+        .unwrap_or_default();
+    dirs.extend(["/usr/sbin", "/sbin", "/usr/bin", "/bin"].map(PathBuf::from));
+    dirs.into_iter()
+        .map(|dir| dir.join(binary))
+        .find(|candidate| candidate.exists() && candidate.is_file())
 }
 
 fn has_cap_net_admin() -> bool {

src/tls_front/emulator.rs

@@ -155,57 +155,35 @@ fn push_fallback_size(sizes: &mut Vec<usize>, size: usize) {
 }
 
 fn fallback_family_app_data_sizes(cached: &CachedTlsData) -> Vec<usize> {
-    if matches!(cached.behavior_profile.source, TlsProfileSource::Rustls)
-        && !cached.app_data_records_sizes.is_empty()
-    {
-        return cached.app_data_records_sizes.clone();
-    }
-
-    let family = fallback_shape_family(cached);
-    let mut remaining = fallback_total_app_data_len(cached);
-    let preferred_chunk = match family {
-        FallbackShapeFamily::NginxLike => 2896,
-        FallbackShapeFamily::BoringSslLike => 1369,
-        FallbackShapeFamily::RustlsLike => 2048,
+    let mut sizes = Vec::with_capacity(1);
+    let size = if matches!(cached.behavior_profile.source, TlsProfileSource::Rustls) {
+        cached
+            .app_data_records_sizes
+            .first()
+            .copied()
+            .unwrap_or_else(|| fallback_total_app_data_len(cached))
+    } else {
+        fallback_total_app_data_len(cached)
     };
-    let split_threshold = match family {
-        FallbackShapeFamily::NginxLike => 4096,
-        FallbackShapeFamily::BoringSslLike => 1536,
-        FallbackShapeFamily::RustlsLike => 3072,
-    };
-
-    if remaining <= split_threshold {
-        return vec![remaining.clamp(MIN_APP_DATA, MAX_APP_DATA)];
-    }
-
-    let mut sizes: Vec<usize> = Vec::new();
-    while remaining > 0 {
-        let chunk = remaining.min(preferred_chunk).min(MAX_APP_DATA);
-        if chunk < MIN_APP_DATA {
-            if let Some(last) = sizes.last_mut() {
-                *last = (*last).saturating_add(chunk).min(MAX_APP_DATA);
-            } else {
-                push_fallback_size(&mut sizes, chunk);
-            }
-            break;
-        }
-        push_fallback_size(&mut sizes, chunk);
-        remaining = remaining.saturating_sub(chunk);
-    }
-
+    push_fallback_size(&mut sizes, size);
     sizes
 }
 
 fn emulated_app_data_sizes(cached: &CachedTlsData) -> Vec<usize> {
     match cached.behavior_profile.source {
         TlsProfileSource::Raw | TlsProfileSource::Merged => {
-            if !cached.behavior_profile.app_data_record_sizes.is_empty() {
-                return cached.behavior_profile.app_data_record_sizes.clone();
+            if let Some(size) = cached.behavior_profile.app_data_record_sizes.first() {
+                return vec![(*size).clamp(MIN_APP_DATA, MAX_APP_DATA)];
             }
-            if !cached.app_data_records_sizes.is_empty() {
-                return cached.app_data_records_sizes.clone();
+            if let Some(size) = cached.app_data_records_sizes.first() {
+                return vec![(*size).clamp(MIN_APP_DATA, MAX_APP_DATA)];
             }
-            return vec![cached.total_app_data_len.max(1024)];
+            return vec![
+                cached
+                    .total_app_data_len
+                    .max(1024)
+                    .clamp(MIN_APP_DATA, MAX_APP_DATA),
+            ];
         }
         TlsProfileSource::Default | TlsProfileSource::Rustls => {
             return fallback_family_app_data_sizes(cached);
@@ -417,7 +395,7 @@ pub fn build_emulated_server_hello(
     alpn: Option<Vec<u8>>,
     new_session_tickets: u8,
 ) -> Vec<u8> {
-    // --- ServerHello ---
+    // ServerHello carries the authenticated digest bytes that the client verifies.
     let extensions = build_profiled_server_hello_extensions(cached, server_key_share);
     let extensions_len = extensions.len() as u16;
 
@@ -449,7 +427,7 @@ pub fn build_emulated_server_hello(
     server_hello.extend_from_slice(&(message.len() as u16).to_be_bytes());
     server_hello.extend_from_slice(&message);
 
-    // --- ChangeCipherSpec ---
+    // ChangeCipherSpec is part of the client-visible TLS shim prefix.
     let change_cipher_spec_count = emulated_change_cipher_spec_count(cached);
     let mut change_cipher_spec = Vec::with_capacity(change_cipher_spec_count * 6);
     for _ in 0..change_cipher_spec_count {
@@ -463,7 +441,8 @@ pub fn build_emulated_server_hello(
         ]);
     }
 
-    // --- ApplicationData (fake encrypted records) ---
+    // Telegram clients authenticate the hello prefix and then expose any later
+    // ApplicationData bytes to the MTProto packet parser.
     let mut sizes = {
         let base_sizes = emulated_app_data_sizes(cached);
         match cached.behavior_profile.source {
@@ -550,8 +529,7 @@ pub fn build_emulated_server_hello(
         app_data.extend_from_slice(&rec);
     }
 
-    // --- Combine ---
-    // Optional NewSessionTicket mimic records (opaque ApplicationData for fingerprint).
+    // Optional NewSessionTicket mimic records are an explicit fingerprint opt-in.
     let mut tickets = Vec::new();
     for ticket_len in emulated_ticket_record_sizes(cached, new_session_tickets, rng) {
         let mut rec = Vec::with_capacity(5 + ticket_len);
@@ -570,7 +548,7 @@ pub fn build_emulated_server_hello(
     response.extend_from_slice(&app_data);
     response.extend_from_slice(&tickets);
 
-    // --- HMAC ---
+    // The digest authenticates the server response bytes emitted by this builder.
     let mut hmac_input = Vec::with_capacity(TLS_DIGEST_LEN + response.len());
     hmac_input.extend_from_slice(client_digest);
     hmac_input.extend_from_slice(&response);
@@ -1062,7 +1040,7 @@ mod tests {
             app_lens.push(record_len);
             pos += 5 + record_len;
         }
-        assert_eq!(app_lens, vec![64, 3905, 537]);
+        assert_eq!(app_lens, vec![64]);
         assert_eq!(pos, response.len());
     }
 }

src/tls_front/tests/emulator_profile_fidelity_security_tests.rs

@@ -106,7 +106,37 @@ fn emulated_server_hello_does_not_emit_profile_ticket_tail_when_disabled() {
     );
 
     let app_records = record_lengths_by_type(&response, TLS_RECORD_APPLICATION);
-    assert_eq!(app_records, vec![1200, 900]);
+    assert_eq!(app_records, vec![1200]);
+}
+
+#[test]
+fn emulated_server_hello_keeps_default_profile_primary_app_data_single() {
+    let mut cached = make_cached();
+    cached.behavior_profile.source = TlsProfileSource::Default;
+    cached.behavior_profile.app_data_record_sizes.clear();
+    cached.behavior_profile.ticket_record_sizes.clear();
+    cached.app_data_records_sizes = vec![2048, 1024];
+    cached.total_app_data_len = 5000;
+    let rng = SecureRandom::new();
+
+    let response = build_emulated_server_hello(
+        b"secret",
+        &[0x85; 32],
+        &[0x86; 16],
+        &cached,
+        false,
+        true,
+        ClientHelloTlsVersion::Tls13,
+        [0x13, 0x01],
+        &test_server_key_share(),
+        &rng,
+        None,
+        0,
+    );
+
+    let app_records = record_lengths_by_type(&response, TLS_RECORD_APPLICATION);
+    assert_eq!(app_records.len(), 1);
+    assert!(app_records[0] >= 64);
 }
 
 #[test]
@@ -130,5 +160,5 @@ fn emulated_server_hello_uses_profile_ticket_lengths_when_enabled() {
     );
 
     let app_records = record_lengths_by_type(&response, TLS_RECORD_APPLICATION);
-    assert_eq!(app_records, vec![1200, 900, 220, 180]);
+    assert_eq!(app_records, vec![1200, 220, 180]);
 }