From a55efe858cec2c4aaee85e4f97d0c6e7484ce928 Mon Sep 17 00:00:00 2001
From: Jeongkyu Shin <inureyes@gmail.com>
Date: Fri, 28 Aug 2026 13:26:54 +0900
Subject: [PATCH 1/5] feat(ssh): enforce RekeyLimit transport policy (#301)

- Replace raw RekeyLimit strings with a checked byte/time policy and preserve OpenSSH's `default none` semantics.
- Apply the effective policy to russh read, write, and time limits while capping data at the backend's nonce-safety ceiling.
- Forward piped stdin for a single raw SSH destination so bounded transfer regressions exercise real transport rekeys without racing multi-host readers.

- `cargo test rekey --lib`
- focused ssh_config parser, resolver, and runtime registry tests
- `cargo check --lib --bins --tests`
- `cargo clippy --lib --bins --tests -- -D warnings`
- `cargo fmt --all -- --check`
- `git diff --check`

Refs #301
---
 src/executor/parallel.rs                      |  26 +-
 src/ssh/client/command.rs                     |  73 ++++-
 src/ssh/ssh_config/mod.rs                     |  36 ++-
 .../ssh_config/parser/options/connection.rs   | 162 +---------
 src/ssh/ssh_config/parser/options/support.rs  |   5 +-
 src/ssh/ssh_config/rekey.rs                   | 303 ++++++++++++++++++
 src/ssh/ssh_config/resolver.rs                |   2 +-
 src/ssh/ssh_config/types.rs                   |   4 +-
 src/ssh/tokio_client/channel_manager.rs       | 125 +++++---
 src/ssh/tokio_client/connection.rs            |  18 +-
 src/ssh/tokio_client/connection_tests.rs      |  37 ++-
 src/ssh/tokio_client/session.rs               |  60 +++-
 12 files changed, 633 insertions(+), 218 deletions(-)
 create mode 100644 src/ssh/ssh_config/rekey.rs

diff --git a/src/executor/parallel.rs b/src/executor/parallel.rs
index 948a23d6..13c0edaf 100644
--- a/src/executor/parallel.rs
+++ b/src/executor/parallel.rs
@@ -1238,6 +1238,10 @@ impl ParallelExecutor {
 
         let semaphore = Arc::new(Semaphore::new(self.max_parallel));
         let raw_stream = output_mode.is_raw_stream();
+        // A single OpenSSH-compatible destination owns piped stdin. Never let
+        // parallel fan-out race multiple readers over the process input, and
+        // leave terminal input to the interactive/PTY path.
+        let forward_stdin = raw_stream && self.nodes.len() == 1 && !std::io::stdin().is_terminal();
         let mut manager = MultiNodeStreamManager::new();
         let mut handles = Vec::new();
         let stdin_is_terminal = std::io::stdin().is_terminal();
@@ -1364,10 +1368,24 @@ impl ParallelExecutor {
                         }
                     }
                 } else {
-                    match client
-                        .connect_and_execute_with_output_streaming(&command, &config, tx.clone())
-                        .await
-                    {
+                    let execution = if forward_stdin {
+                        client
+                            .connect_and_execute_with_output_streaming_and_stdin(
+                                &command,
+                                &config,
+                                tx.clone(),
+                            )
+                            .await
+                    } else {
+                        client
+                            .connect_and_execute_with_output_streaming(
+                                &command,
+                                &config,
+                                tx.clone(),
+                            )
+                            .await
+                    };
+                    match execution {
                         Ok(exit_status) => {
                             tracing::debug!(
                                 "Command completed for {}: exit code {}",
diff --git a/src/ssh/client/command.rs b/src/ssh/client/command.rs
index 80c7e7ee..ea715362 100644
--- a/src/ssh/client/command.rs
+++ b/src/ssh/client/command.rs
@@ -254,6 +254,28 @@ impl SshClient {
         command: &str,
         config: &ConnectionConfig<'_>,
         output_sender: Sender<CommandOutput>,
+    ) -> Result<u32> {
+        self.connect_and_execute_with_output_streaming_inner(command, config, output_sender, false)
+            .await
+    }
+
+    /// Execute a command with byte-transparent stdout/stderr and piped stdin.
+    pub async fn connect_and_execute_with_output_streaming_and_stdin(
+        &mut self,
+        command: &str,
+        config: &ConnectionConfig<'_>,
+        output_sender: Sender<CommandOutput>,
+    ) -> Result<u32> {
+        self.connect_and_execute_with_output_streaming_inner(command, config, output_sender, true)
+            .await
+    }
+
+    async fn connect_and_execute_with_output_streaming_inner(
+        &mut self,
+        command: &str,
+        config: &ConnectionConfig<'_>,
+        output_sender: Sender<CommandOutput>,
+        forward_stdin: bool,
     ) -> Result<u32> {
         tracing::debug!("Connecting to {}:{}", self.host, self.port);
 
@@ -305,6 +327,7 @@ impl SshClient {
                 config.timeout_seconds,
                 output_sender,
                 config.session_policy,
+                forward_stdin,
             )
             .await
         }
@@ -321,14 +344,22 @@ impl SshClient {
         command: &str,
         output_sender: Sender<CommandOutput>,
         session_policy: Option<&crate::ssh::SessionPolicy>,
+        forward_stdin: bool,
     ) -> Result<u32, crate::ssh::tokio_client::Error> {
         match session_policy {
+            Some(policy) if forward_stdin => {
+                client
+                    .execute_session_streaming_with_stdin(policy, output_sender)
+                    .await
+            }
             Some(policy) => {
                 client
                     .execute_session_streaming(policy, output_sender)
                     .await
             }
-            None => client.execute_streaming(command, output_sender).await,
+            None => {
+                Self::execute_streaming_once(client, command, output_sender, forward_stdin).await
+            }
         }
     }
 
@@ -340,12 +371,19 @@ impl SshClient {
         timeout_seconds: Option<u64>,
         output_sender: Sender<CommandOutput>,
         session_policy: Option<&crate::ssh::SessionPolicy>,
+        forward_stdin: bool,
     ) -> Result<u32> {
         if let Some(timeout_secs) = timeout_seconds {
             if timeout_secs == 0 {
                 // No timeout (unlimited)
                 tracing::debug!("Executing command with streaming, no timeout (unlimited)");
-                Self::execute_resolved_session_streaming(client, command, output_sender, session_policy)
+                Self::execute_resolved_session_streaming(
+                    client,
+                    command,
+                    output_sender,
+                    session_policy,
+                    forward_stdin,
+                )
                     .await
                     .with_context(|| format!("Failed to execute command '{}' on {}:{}. The SSH connection was successful but the command could not be executed.", command, self.host, self.port))
             } else {
@@ -357,7 +395,13 @@ impl SshClient {
                 );
                 tokio::time::timeout(
                     command_timeout,
-                    Self::execute_resolved_session_streaming(client, command, output_sender, session_policy)
+                    Self::execute_resolved_session_streaming(
+                        client,
+                        command,
+                        output_sender,
+                        session_policy,
+                        forward_stdin,
+                    )
                 )
                 .await
                 .with_context(|| format!("Command execution timeout: The command '{}' did not complete within {} seconds on {}:{}", command, timeout_secs, self.host, self.port))?
@@ -369,7 +413,13 @@ impl SshClient {
             tracing::debug!("Executing command with streaming, default timeout of 300 seconds");
             tokio::time::timeout(
                 command_timeout,
-                Self::execute_resolved_session_streaming(client, command, output_sender, session_policy)
+                Self::execute_resolved_session_streaming(
+                    client,
+                    command,
+                    output_sender,
+                    session_policy,
+                    forward_stdin,
+                )
             )
             .await
             .with_context(|| format!("Command execution timeout: The command '{}' did not complete within 5 minutes on {}:{}", command, self.host, self.port))?
@@ -377,6 +427,21 @@ impl SshClient {
         }
     }
 
+    async fn execute_streaming_once(
+        client: &crate::ssh::tokio_client::Client,
+        command: &str,
+        output_sender: Sender<CommandOutput>,
+        forward_stdin: bool,
+    ) -> Result<u32, crate::ssh::tokio_client::Error> {
+        if forward_stdin {
+            client
+                .execute_streaming_with_stdin(command, output_sender)
+                .await
+        } else {
+            client.execute_streaming(command, output_sender).await
+        }
+    }
+
     /// Execute a command with sudo password support and streaming output.
     ///
     /// This method handles automatic sudo password injection when sudo prompts are detected
diff --git a/src/ssh/ssh_config/mod.rs b/src/ssh/ssh_config/mod.rs
index eb8ecece..1d619c56 100644
--- a/src/ssh/ssh_config/mod.rs
+++ b/src/ssh/ssh_config/mod.rs
@@ -30,6 +30,7 @@ mod match_directive;
 mod parser;
 mod path;
 mod pattern;
+mod rekey;
 mod resolver;
 #[cfg(test)]
 mod resolver_tests;
@@ -40,6 +41,9 @@ mod types;
 
 // Re-export public types
 pub use ip_qos::{IpQosParseError, IpQosPolicy, IpQosValue};
+pub use rekey::{
+    RUSSH_REKEY_BYTE_CEILING, RekeyDataLimit, RekeyLimit, RekeyLimitParseError, RekeyTimeLimit,
+};
 pub use types::SshHostConfig;
 
 /// SSH configuration parser and resolver
@@ -635,7 +639,13 @@ Host backup-server
                 bulk: IpQosValue::None,
             })
         );
-        assert_eq!(host1.rekey_limit, Some("1G 1h".to_string()));
+        assert_eq!(
+            host1.rekey_limit,
+            Some(RekeyLimit {
+                data: RekeyDataLimit::Bytes(1 << 30),
+                time: RekeyTimeLimit::Seconds(3_600),
+            })
+        );
 
         // Verify backup-server config
         let host2 = &config.hosts[1];
@@ -647,7 +657,13 @@ Host backup-server
                 bulk: IpQosValue::Class(0x48),
             })
         );
-        assert_eq!(host2.rekey_limit, Some("default none".to_string()));
+        assert_eq!(
+            host2.rekey_limit,
+            Some(RekeyLimit {
+                data: RekeyDataLimit::Default,
+                time: RekeyTimeLimit::None,
+            })
+        );
     }
 
     #[test]
@@ -740,7 +756,13 @@ Host web1.example.com
         );
 
         // RekeyLimit should be from web1.example.com (most specific)
-        assert_eq!(host_config.rekey_limit, Some("1G 2h".to_string()));
+        assert_eq!(
+            host_config.rekey_limit,
+            Some(RekeyLimit {
+                data: RekeyDataLimit::Bytes(1 << 30),
+                time: RekeyTimeLimit::Seconds(7_200),
+            })
+        );
 
         // ForwardX11Timeout should be from *.example.com
         assert_eq!(host_config.forward_x11_timeout, Some("30m".to_string()));
@@ -917,7 +939,13 @@ Host test
     RekeyLimit 1G 1h
 "#;
         let config = SshConfig::parse(config_content).unwrap();
-        assert_eq!(config.hosts[0].rekey_limit, Some("1G 1h".to_string()));
+        assert_eq!(
+            config.hosts[0].rekey_limit,
+            Some(RekeyLimit {
+                data: RekeyDataLimit::Bytes(1 << 30),
+                time: RekeyTimeLimit::Seconds(3_600),
+            })
+        );
 
         // Test ForwardX11Timeout - should reject invalid format
         let config_content = r#"
diff --git a/src/ssh/ssh_config/parser/options/connection.rs b/src/ssh/ssh_config/parser/options/connection.rs
index 72c70572..5e839142 100644
--- a/src/ssh/ssh_config/parser/options/connection.rs
+++ b/src/ssh/ssh_config/parser/options/connection.rs
@@ -17,7 +17,7 @@
 //! Handles connection-related configuration options including keepalive
 //! settings, timeouts, compression, and network settings.
 
-use crate::ssh::ssh_config::IpQosPolicy;
+use crate::ssh::ssh_config::{IpQosPolicy, RekeyLimit};
 use crate::ssh::ssh_config::parser::helpers::parse_yes_no;
 use crate::ssh::ssh_config::types::SshHostConfig;
 use anyhow::{Context, Result};
@@ -177,162 +177,20 @@ pub(super) fn parse_connection_option(
             if args.is_empty() {
                 anyhow::bail!("RekeyLimit requires a value at line {line_number}");
             }
-            // RekeyLimit can have one or two values (data limit and time limit)
-            // Format: <data> [<time>]
-            // Data: default, none, or number with optional suffix (K/M/G/T)
-            // Time: none or number with optional suffix (s/m/h/d/w)
-
-            if args.len() > 2 {
+            if args.iter().any(|value| value.len() > 20) || args.join(" ").len() > 50 {
                 anyhow::bail!(
-                    "RekeyLimit at line {} accepts at most 2 values (data and time), got {}",
-                    line_number,
-                    args.len()
+                    "RekeyLimit value at line {line_number} is too long (max 50 characters total)"
                 );
             }
-
-            // Validate data limit (first argument)
-            let data_limit = &args[0];
-            if data_limit != "default" && data_limit != "none" {
-                // Parse size with optional suffix
-                let (number_part, _suffix, multiplier) =
-                    if let Some(stripped) = data_limit.strip_suffix(&['K', 'k'][..]) {
-                        (stripped, "K", 1024u64)
-                    } else if let Some(stripped) = data_limit.strip_suffix(&['M', 'm'][..]) {
-                        (stripped, "M", 1024u64 * 1024)
-                    } else if let Some(stripped) = data_limit.strip_suffix(&['G', 'g'][..]) {
-                        (stripped, "G", 1024u64 * 1024 * 1024)
-                    } else if let Some(stripped) = data_limit.strip_suffix(&['T', 't'][..]) {
-                        (stripped, "T", 1024u64 * 1024 * 1024 * 1024)
-                    } else {
-                        // Plain number (bytes)
-                        (data_limit.as_str(), "", 1u64)
-                    };
-
-                match number_part.parse::<u64>() {
-                    Ok(num) => {
-                        // Check for overflow when applying multiplier
-                        if let Some(total) = num.checked_mul(multiplier) {
-                            // Warn if rekey limit is very large (> 1TB)
-                            if total > 1024u64 * 1024 * 1024 * 1024 {
-                                tracing::warn!(
-                                    "RekeyLimit data limit '{}' at line {} is very large ({}TB). \
-                                     This may not be effective for security",
-                                    data_limit,
-                                    line_number,
-                                    total / (1024u64 * 1024 * 1024 * 1024)
-                                );
-                            }
-                            // Warn if rekey limit is very small (< 1KB)
-                            if total < 1024 && total != 0 {
-                                tracing::warn!(
-                                    "RekeyLimit data limit '{}' at line {} is very small ({} bytes). \
-                                     This may cause frequent rekeying",
-                                    data_limit,
-                                    line_number,
-                                    total
-                                );
-                            }
-                        } else {
-                            anyhow::bail!(
-                                "RekeyLimit data limit '{data_limit}' at line {line_number} would overflow. \
-                                 Please use a smaller value"
-                            );
-                        }
-                    }
-                    Err(_) => {
-                        anyhow::bail!(
-                            "RekeyLimit data limit '{data_limit}' at line {line_number} is invalid. \
-                             Use 'default', 'none', or a number with optional suffix (K/M/G/T)"
-                        );
-                    }
-                }
-
-                // Prevent absurdly long input strings
-                if data_limit.len() > 20 {
-                    anyhow::bail!(
-                        "RekeyLimit data limit at line {line_number} is too long (max 20 characters)"
-                    );
-                }
-            }
-
-            // Validate time limit (second argument, if present)
-            if args.len() > 1 {
-                let time_limit = &args[1];
-                if time_limit != "none" {
-                    // Parse time with optional suffix
-                    let (number_part, _suffix, multiplier) =
-                        if let Some(stripped) = time_limit.strip_suffix(&['s', 'S'][..]) {
-                            (stripped, "s", 1u64)
-                        } else if let Some(stripped) = time_limit.strip_suffix(&['m', 'M'][..]) {
-                            (stripped, "m", 60u64)
-                        } else if let Some(stripped) = time_limit.strip_suffix(&['h', 'H'][..]) {
-                            (stripped, "h", 3600u64)
-                        } else if let Some(stripped) = time_limit.strip_suffix(&['d', 'D'][..]) {
-                            (stripped, "d", 86400u64)
-                        } else if let Some(stripped) = time_limit.strip_suffix(&['w', 'W'][..]) {
-                            (stripped, "w", 604800u64)
-                        } else {
-                            // Plain number (seconds)
-                            (time_limit.as_str(), "", 1u64)
-                        };
-
-                    match number_part.parse::<u64>() {
-                        Ok(num) => {
-                            // Check for overflow when applying multiplier
-                            if let Some(total_seconds) = num.checked_mul(multiplier) {
-                                // Warn if rekey time is very long (> 1 week)
-                                if total_seconds > 604800 {
-                                    tracing::warn!(
-                                        "RekeyLimit time limit '{}' at line {} is very long ({} days). \
-                                         This may reduce security",
-                                        time_limit,
-                                        line_number,
-                                        total_seconds / 86400
-                                    );
-                                }
-                                // Warn if rekey time is very short (< 60 seconds)
-                                if total_seconds < 60 && total_seconds != 0 {
-                                    tracing::warn!(
-                                        "RekeyLimit time limit '{}' at line {} is very short ({} seconds). \
-                                         This may cause frequent rekeying",
-                                        time_limit,
-                                        line_number,
-                                        total_seconds
-                                    );
-                                }
-                            } else {
-                                anyhow::bail!(
-                                    "RekeyLimit time limit '{time_limit}' at line {line_number} would overflow. \
-                                     Please use a smaller value"
-                                );
-                            }
-                        }
-                        Err(_) => {
-                            anyhow::bail!(
-                                "RekeyLimit time limit '{time_limit}' at line {line_number} is invalid. \
-                                 Use 'none' or a number with optional suffix (s/m/h/d/w)"
-                            );
-                        }
-                    }
-
-                    // Prevent absurdly long input strings
-                    if time_limit.len() > 20 {
-                        anyhow::bail!(
-                            "RekeyLimit time limit at line {line_number} is too long (max 20 characters)"
-                        );
-                    }
-                }
-            }
-
-            // Limit total length to prevent memory exhaustion
-            let combined = args.join(" ");
-            if combined.len() > 50 {
-                anyhow::bail!(
-                    "RekeyLimit value at line {line_number} is too long (max 50 characters total)"
+            let policy = RekeyLimit::parse(args)
+                .with_context(|| format!("Invalid RekeyLimit value at line {line_number}"))?;
+            if policy.data_is_capped() {
+                tracing::warn!(
+                    "RekeyLimit data policy at line {line_number} exceeds russh's 1 GiB \
+                     nonce-safety ceiling and will be capped"
                 );
             }
-
-            host.rekey_limit = Some(combined);
+            host.rekey_limit = Some(policy);
         }
         _ => unreachable!("Unexpected keyword in parse_connection_option: {}", keyword),
     }
diff --git a/src/ssh/ssh_config/parser/options/support.rs b/src/ssh/ssh_config/parser/options/support.rs
index bf0075ad..3ad6a4ed 100644
--- a/src/ssh/ssh_config/parser/options/support.rs
+++ b/src/ssh/ssh_config/parser/options/support.rs
@@ -195,7 +195,7 @@ pub(super) const ACCEPTED_KEYWORDS: &[(&str, &str, KeywordSupport)] = &[
     ("bindaddress", "bindaddress", Runtime(Transport)),
     ("bindinterface", "bindinterface", Runtime(Transport)),
     ("ipqos", "ipqos", Runtime(Transport)),
-    ("rekeylimit", "rekeylimit", Delegated(301)),
+    ("rekeylimit", "rekeylimit", Runtime(Transport)),
     ("proxyjump", "proxyjump", Runtime(Proxy)),
     ("proxycommand", "proxycommand", Runtime(Proxy)),
     ("proxyusefdpass", "proxyusefdpass", Runtime(Proxy)),
@@ -323,6 +323,7 @@ mod tests {
             ("bindaddress", Transport),
             ("bindinterface", Transport),
             ("ipqos", Transport),
+            ("rekeylimit", Transport),
             ("proxyjump", Proxy),
             ("proxycommand", Proxy),
             ("proxyusefdpass", Proxy),
@@ -353,7 +354,7 @@ mod tests {
 
     #[test]
     fn first_wave_delegations_match_the_split_issue_dag() {
-        let expected = HashMap::from([("rekeylimit", 301)]);
+        let expected = HashMap::new();
         let delegated = ACCEPTED_KEYWORDS
             .iter()
             .filter_map(|(keyword, canonical, support)| match support {
diff --git a/src/ssh/ssh_config/rekey.rs b/src/ssh/ssh_config/rekey.rs
new file mode 100644
index 00000000..935e59d1
--- /dev/null
+++ b/src/ssh/ssh_config/rekey.rs
@@ -0,0 +1,303 @@
+// Copyright 2025 Lablup Inc. and Jeongkyu Shin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Typed `RekeyLimit` parsing and conversion to russh transport limits.
+
+use std::time::Duration;
+
+use thiserror::Error;
+
+/// russh's hard per-direction byte ceiling.
+///
+/// `russh::Limits::new` asserts that neither byte limit exceeds 1 GiB to
+/// prevent nonce reuse. bssh therefore caps larger explicit values instead of
+/// claiming an unlimited transport that the backend cannot provide.
+pub const RUSSH_REKEY_BYTE_CEILING: u64 = 1 << 30;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum RekeyDataLimit {
+    #[default]
+    Default,
+    Bytes(u64),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum RekeyTimeLimit {
+    #[default]
+    Default,
+    None,
+    Seconds(u64),
+}
+
+/// Effective ssh_config rekey policy before applying backend safety limits.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub struct RekeyLimit {
+    pub data: RekeyDataLimit,
+    pub time: RekeyTimeLimit,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Error)]
+pub enum RekeyLimitParseError {
+    #[error("a data limit is required")]
+    MissingDataLimit,
+    #[error("at most a data limit and an optional time limit are accepted")]
+    TooManyValues,
+    #[error("invalid data limit '{0}'")]
+    InvalidDataLimit(String),
+    #[error("data limit '{0}' overflows")]
+    DataLimitOverflow(String),
+    #[error("invalid time limit '{0}'")]
+    InvalidTimeLimit(String),
+    #[error("time limit '{0}' overflows")]
+    TimeLimitOverflow(String),
+}
+
+impl RekeyLimit {
+    pub fn parse(values: &[String]) -> Result<Self, RekeyLimitParseError> {
+        let Some(data) = values.first() else {
+            return Err(RekeyLimitParseError::MissingDataLimit);
+        };
+        if values.len() > 2 {
+            return Err(RekeyLimitParseError::TooManyValues);
+        }
+
+        Ok(Self {
+            data: parse_data_limit(data)?,
+            time: values
+                .get(1)
+                .map_or(Ok(RekeyTimeLimit::Default), |value| parse_time_limit(value))?,
+        })
+    }
+
+    /// Whether the requested data policy must be reduced to russh's hard
+    /// nonce-safety ceiling.
+    #[must_use]
+    pub fn data_is_capped(self) -> bool {
+        matches!(self.data, RekeyDataLimit::Bytes(bytes) if bytes > RUSSH_REKEY_BYTE_CEILING)
+    }
+
+    /// Convert to safe russh limits without panicking or silently disabling
+    /// byte-based rekeying.
+    #[must_use]
+    pub fn to_russh_limits(self) -> russh::Limits {
+        let defaults = russh::Limits::default();
+        let requested_bytes = match self.data {
+            RekeyDataLimit::Default => defaults.rekey_write_limit as u64,
+            RekeyDataLimit::Bytes(bytes) => bytes,
+        };
+        let effective_bytes = requested_bytes.min(RUSSH_REKEY_BYTE_CEILING) as usize;
+        let effective_time = match self.time {
+            // OpenSSH's effective default is `RekeyLimit default none`, not
+            // russh's one-hour backend default.
+            RekeyTimeLimit::Default | RekeyTimeLimit::None => Duration::MAX,
+            RekeyTimeLimit::Seconds(seconds) => Duration::from_secs(seconds),
+        };
+
+        russh::Limits::new(effective_bytes, effective_bytes, effective_time)
+    }
+}
+
+fn parse_data_limit(value: &str) -> Result<RekeyDataLimit, RekeyLimitParseError> {
+    if value.eq_ignore_ascii_case("default") {
+        return Ok(RekeyDataLimit::Default);
+    }
+    let (number, multiplier) = split_suffix(
+        value,
+        &[
+            ("k", 1 << 10),
+            ("m", 1 << 20),
+            ("g", 1 << 30),
+            ("t", 1 << 40),
+        ],
+    );
+    let parsed = number
+        .parse::<u64>()
+        .map_err(|_| RekeyLimitParseError::InvalidDataLimit(value.to_string()))?;
+    if parsed == 0 {
+        return Ok(RekeyDataLimit::Default);
+    }
+    let bytes = parsed
+        .checked_mul(multiplier)
+        .ok_or_else(|| RekeyLimitParseError::DataLimitOverflow(value.to_string()))?;
+    if bytes < 16 {
+        return Err(RekeyLimitParseError::InvalidDataLimit(value.to_string()));
+    }
+    Ok(RekeyDataLimit::Bytes(bytes))
+}
+
+fn parse_time_limit(value: &str) -> Result<RekeyTimeLimit, RekeyLimitParseError> {
+    if value.eq_ignore_ascii_case("none") {
+        return Ok(RekeyTimeLimit::None);
+    }
+
+    let (number, multiplier) = split_suffix(
+        value,
+        &[
+            ("s", 1),
+            ("m", 60),
+            ("h", 3_600),
+            ("d", 86_400),
+            ("w", 604_800),
+        ],
+    );
+    let parsed = number
+        .parse::<u64>()
+        .map_err(|_| RekeyLimitParseError::InvalidTimeLimit(value.to_string()))?;
+    if parsed == 0 {
+        return Ok(RekeyTimeLimit::Default);
+    }
+    parsed
+        .checked_mul(multiplier)
+        .map(RekeyTimeLimit::Seconds)
+        .ok_or_else(|| RekeyLimitParseError::TimeLimitOverflow(value.to_string()))
+}
+
+fn split_suffix<'a>(value: &'a str, suffixes: &[(&str, u64)]) -> (&'a str, u64) {
+    suffixes
+        .iter()
+        .find_map(|(suffix, multiplier)| {
+            value
+                .get(value.len().saturating_sub(suffix.len())..)
+                .filter(|ending| ending.eq_ignore_ascii_case(suffix))
+                .and_then(|_| value.get(..value.len().saturating_sub(suffix.len())))
+                .map(|number| (number, *multiplier))
+        })
+        .unwrap_or((value, 1))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn values(parts: &[&str]) -> Vec<String> {
+        parts.iter().map(|part| (*part).to_string()).collect()
+    }
+
+    #[test]
+    fn parses_default_none_and_suffixes() {
+        assert_eq!(
+            RekeyLimit::parse(&values(&["default"])),
+            Ok(RekeyLimit::default())
+        );
+        assert_eq!(
+            RekeyLimit::parse(&values(&["default", "none"])),
+            Ok(RekeyLimit {
+                data: RekeyDataLimit::Default,
+                time: RekeyTimeLimit::None,
+            })
+        );
+        assert_eq!(
+            RekeyLimit::parse(&values(&["16M", "2h"])),
+            Ok(RekeyLimit {
+                data: RekeyDataLimit::Bytes(16 << 20),
+                time: RekeyTimeLimit::Seconds(7_200),
+            })
+        );
+        assert_eq!(
+            RekeyLimit::parse(&values(&["2t", "1W"])),
+            Ok(RekeyLimit {
+                data: RekeyDataLimit::Bytes(2 << 40),
+                time: RekeyTimeLimit::Seconds(604_800),
+            })
+        );
+    }
+
+    #[test]
+    fn numeric_zero_uses_backend_defaults() {
+        assert_eq!(
+            RekeyLimit::parse(&values(&["0", "0"])),
+            Ok(RekeyLimit::default())
+        );
+    }
+
+    #[test]
+    fn rejects_missing_extra_invalid_and_overflow_values() {
+        assert_eq!(
+            RekeyLimit::parse(&[]),
+            Err(RekeyLimitParseError::MissingDataLimit)
+        );
+        assert_eq!(
+            RekeyLimit::parse(&values(&["1M", "1h", "extra"])),
+            Err(RekeyLimitParseError::TooManyValues)
+        );
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["invalid"])),
+            Err(RekeyLimitParseError::InvalidDataLimit(_))
+        ));
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["none"])),
+            Err(RekeyLimitParseError::InvalidDataLimit(_))
+        ));
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["15"])),
+            Err(RekeyLimitParseError::InvalidDataLimit(_))
+        ));
+        assert_eq!(
+            RekeyLimit::parse(&values(&["16"])),
+            Ok(RekeyLimit {
+                data: RekeyDataLimit::Bytes(16),
+                time: RekeyTimeLimit::Default,
+            })
+        );
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["1M", "forever"])),
+            Err(RekeyLimitParseError::InvalidTimeLimit(_))
+        ));
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["18446744073709551615T"])),
+            Err(RekeyLimitParseError::DataLimitOverflow(_))
+        ));
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["1M", "18446744073709551615w"])),
+            Err(RekeyLimitParseError::TimeLimitOverflow(_))
+        ));
+    }
+
+    #[test]
+    fn conversion_preserves_safe_values_and_caps_unsafe_data_requests() {
+        let small = RekeyLimit {
+            data: RekeyDataLimit::Bytes(4_096),
+            time: RekeyTimeLimit::Seconds(30),
+        };
+        let small_limits = small.to_russh_limits();
+        assert_eq!(small_limits.rekey_write_limit, 4_096);
+        assert_eq!(small_limits.rekey_read_limit, 4_096);
+        assert_eq!(small_limits.rekey_time_limit, Duration::from_secs(30));
+        assert!(!small.data_is_capped());
+
+        let policy = RekeyLimit {
+            data: RekeyDataLimit::Bytes(RUSSH_REKEY_BYTE_CEILING + 1),
+            time: RekeyTimeLimit::Default,
+        };
+        let limits = policy.to_russh_limits();
+        assert_eq!(limits.rekey_write_limit, 1 << 30);
+        assert_eq!(limits.rekey_read_limit, 1 << 30);
+        assert!(policy.data_is_capped());
+    }
+
+    #[test]
+    fn default_and_time_none_preserve_the_byte_safety_ceiling() {
+        let limits = RekeyLimit::default().to_russh_limits();
+        assert_eq!(limits.rekey_write_limit, 1 << 30);
+        assert_eq!(limits.rekey_read_limit, 1 << 30);
+        assert_eq!(limits.rekey_time_limit, Duration::MAX);
+
+        let explicit_none = RekeyLimit {
+            data: RekeyDataLimit::Default,
+            time: RekeyTimeLimit::None,
+        }
+        .to_russh_limits();
+        assert_eq!(explicit_none.rekey_time_limit, Duration::MAX);
+    }
+}
diff --git a/src/ssh/ssh_config/resolver.rs b/src/ssh/ssh_config/resolver.rs
index d20dc906..2bc0be2b 100644
--- a/src/ssh/ssh_config/resolver.rs
+++ b/src/ssh/ssh_config/resolver.rs
@@ -360,7 +360,7 @@ pub(super) fn merge_host_config(base: &mut SshHostConfig, overlay: &SshHostConfi
         base.ipqos = overlay.ipqos;
     }
     if base.rekey_limit.is_none() && overlay.rekey_limit.is_some() {
-        base.rekey_limit = overlay.rekey_limit.clone();
+        base.rekey_limit = overlay.rekey_limit;
     }
     // X11 forwarding
     if base.forward_x11_timeout.is_none() && overlay.forward_x11_timeout.is_some() {
diff --git a/src/ssh/ssh_config/types.rs b/src/ssh/ssh_config/types.rs
index 43a6c68d..8c999631 100644
--- a/src/ssh/ssh_config/types.rs
+++ b/src/ssh/ssh_config/types.rs
@@ -20,7 +20,7 @@ use std::path::PathBuf;
 
 use crate::forwarding::ForwardingDirective;
 
-use super::IpQosPolicy;
+use super::{IpQosPolicy, RekeyLimit};
 
 /// Configuration block type
 #[derive(Debug, Clone, PartialEq)]
@@ -126,7 +126,7 @@ pub struct SshHostConfig {
     // Network & connection
     pub bind_interface: Option<String>,
     pub ipqos: Option<IpQosPolicy>,
-    pub rekey_limit: Option<String>,
+    pub rekey_limit: Option<RekeyLimit>,
     // X11 forwarding
     pub forward_x11_timeout: Option<String>,
     pub forward_x11_trusted: Option<bool>,
diff --git a/src/ssh/tokio_client/channel_manager.rs b/src/ssh/tokio_client/channel_manager.rs
index f8f0a134..ae5236c8 100644
--- a/src/ssh/tokio_client/channel_manager.rs
+++ b/src/ssh/tokio_client/channel_manager.rs
@@ -25,6 +25,7 @@ use russh::Channel;
 use russh::client::Msg;
 use std::io;
 use std::net::SocketAddr;
+use tokio::io::{AsyncRead, AsyncReadExt};
 use tokio::sync::mpsc::{Receiver, Sender, channel};
 use tokio::task::JoinHandle;
 
@@ -87,6 +88,27 @@ async fn forward_command_output(sender: &Sender<CommandOutput>, output: CommandO
     }
 }
 
+async fn handle_command_message(
+    sender: &Sender<CommandOutput>,
+    message: russh::ChannelMsg,
+    output_receiver_open: &mut bool,
+    result: &mut Option<u32>,
+) {
+    match message {
+        russh::ChannelMsg::Data { data } if *output_receiver_open => {
+            *output_receiver_open =
+                forward_command_output(sender, CommandOutput::StdOut(data)).await;
+        }
+        russh::ChannelMsg::ExtendedData { data, ext: 1 } if *output_receiver_open => {
+            *output_receiver_open =
+                forward_command_output(sender, CommandOutput::StdErr(data)).await;
+        }
+        // An exit status does not imply that all channel data has arrived.
+        russh::ChannelMsg::ExitStatus { exit_status } => *result = Some(exit_status),
+        _ => {}
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct DirectTcpipRequestTarget {
     host: String,
@@ -301,8 +323,6 @@ impl Client {
     /// Returns only the exit status of the command. Stdout and stderr are streamed
     /// through the sender channel.
     ///
-    /// Make sure your commands don't read from stdin and exit after bounded time.
-    ///
     /// Can be called multiple times, but every invocation is a new shell context.
     /// Thus `cd`, setting variables and alike have no effect on future invocations.
     ///
@@ -328,6 +348,30 @@ impl Client {
         command: &str,
         sender: Sender<CommandOutput>,
     ) -> Result<u32, super::Error> {
+        self.execute_streaming_with_input(command, sender, tokio::io::empty())
+            .await
+    }
+
+    /// Execute a remote command while forwarding the process stdin with
+    /// bounded reads and SSH channel backpressure.
+    pub async fn execute_streaming_with_stdin(
+        &self,
+        command: &str,
+        sender: Sender<CommandOutput>,
+    ) -> Result<u32, super::Error> {
+        self.execute_streaming_with_input(command, sender, tokio::io::stdin())
+            .await
+    }
+
+    async fn execute_streaming_with_input<R>(
+        &self,
+        command: &str,
+        sender: Sender<CommandOutput>,
+        mut input: R,
+    ) -> Result<u32, super::Error>
+    where
+        R: AsyncRead + Unpin,
+    {
         let execution = async {
             // Sanitize command to prevent injection attacks
             let sanitized_command = crate::utils::sanitize_command(command)
@@ -350,56 +394,51 @@ impl Client {
 
             let mut result: Option<u32> = None;
             let mut output_receiver_open = true;
-
-            // While the channel has messages...
-            while let Some(msg) = channel.wait().await {
-                match msg {
-                    // If we get data, send it to the streaming channel
-                    // Note: We must clone the data here because russh owns it and will reuse the buffer
-                    russh::ChannelMsg::Data { ref data } => {
-                        if output_receiver_open {
-                            output_receiver_open = forward_command_output(
-                                &sender,
-                                CommandOutput::StdOut(data.clone()),
-                            )
-                            .await;
+            let mut input_open = true;
+            let mut input_buffer = [0_u8; 32 * 1024];
+
+            loop {
+                if input_open {
+                    tokio::select! {
+                        read = input.read(&mut input_buffer) => {
+                            match read.map_err(super::Error::IoError)? {
+                                0 => {
+                                    channel.eof().await?;
+                                    input_open = false;
+                                }
+                                count => channel.data(&input_buffer[..count]).await?,
+                            }
                         }
-                    }
-                    russh::ChannelMsg::ExtendedData { ref data, ext: 1 } => {
-                        if output_receiver_open {
-                            output_receiver_open = forward_command_output(
+                        message = channel.wait() => {
+                            let Some(message) = message else {
+                                break;
+                            };
+                            handle_command_message(
                                 &sender,
-                                CommandOutput::StdErr(data.clone()),
-                            )
-                            .await;
+                                message,
+                                &mut output_receiver_open,
+                                &mut result,
+                            ).await;
                         }
                     }
-
-                    // If we get an exit code report, store it, but crucially don't
-                    // assume this message means end of communications. The data might
-                    // not be finished yet!
-                    russh::ChannelMsg::ExitStatus { exit_status } => result = Some(exit_status),
-
-                    // We SHOULD get this EOF message, but 4254 sec 5.3 also permits
-                    // the channel to close without it being sent. And sometimes this
-                    // message can even precede the Data message, so don't handle it
-                    // russh::ChannelMsg::Eof => break,
-                    _ => {}
+                } else {
+                    let Some(message) = channel.wait().await else {
+                        break;
+                    };
+                    handle_command_message(
+                        &sender,
+                        message,
+                        &mut output_receiver_open,
+                        &mut result,
+                    )
+                    .await;
                 }
             }
 
-            // Drop sender to signal completion to receiver
-            // This is critical: dropping the sender causes receiver.recv() to return None,
-            // allowing the background task to finish collecting any remaining buffered data
+            // Dropping the sender signals the output collector after the SSH channel is drained.
             drop(sender);
 
-            // If we received an exit code, report it back
-            if let Some(result) = result {
-                Ok(result)
-            // Otherwise, report an error
-            } else {
-                Err(self.session_error_or(super::Error::CommandDidntExit))
-            }
+            result.ok_or_else(|| self.session_error_or(super::Error::CommandDidntExit))
         }
         .await;
         self.flush_hostkey_updates().await;
diff --git a/src/ssh/tokio_client/connection.rs b/src/ssh/tokio_client/connection.rs
index a6b557a0..ba8f44eb 100644
--- a/src/ssh/tokio_client/connection.rs
+++ b/src/ssh/tokio_client/connection.rs
@@ -36,7 +36,7 @@ use super::proxy_command::{
 };
 use crate::forwarding::remote::RemoteForwardRegistry;
 use crate::forwarding::{ForwardingDirective, ForwardingPlan, ForwardingRuntime};
-use crate::ssh::ssh_config::{IpQosPolicy, IpQosValue};
+use crate::ssh::ssh_config::{IpQosPolicy, IpQosValue, RekeyLimit};
 use crate::ssh::{SessionPurpose, SshConfig};
 
 /// Default keepalive interval in seconds.
@@ -160,6 +160,8 @@ pub struct SshConnectionConfig {
     pub auth_policy: SshAuthenticationPolicy,
     /// Forwarding directives resolved for this destination.
     pub forwarding_plan: ForwardingPlan,
+    /// Effective byte/time policy for SSH transport key exchange.
+    pub rekey_limit: RekeyLimit,
 }
 
 impl Default for SshConnectionConfig {
@@ -192,6 +194,7 @@ impl Default for SshConnectionConfig {
             pubkey_accepted_algorithms: None,
             auth_policy: SshAuthenticationPolicy::default(),
             forwarding_plan: ForwardingPlan::default(),
+            rekey_limit: RekeyLimit::default(),
         }
     }
 }
@@ -553,6 +556,10 @@ impl SshConnectionConfigResolver {
                 .unwrap_or(false),
             address_family,
         };
+        let rekey_limit = host_config
+            .as_ref()
+            .and_then(|config| config.rekey_limit)
+            .unwrap_or_default();
         SshConnectionConfig::new()
             .with_keepalive_interval(if keepalive_interval == 0 {
                 None
@@ -586,6 +593,7 @@ impl SshConnectionConfigResolver {
             .with_pubkey_accepted_algorithms(pubkey_accepted_algorithms)
             .with_auth_policy(auth_policy)
             .with_forwarding_plan(forwarding_plan)
+            .with_rekey_limit(rekey_limit)
     }
 
     fn resolve_proxy_mode(
@@ -830,6 +838,13 @@ impl SshConnectionConfig {
         self
     }
 
+    /// Set the effective ssh_config `RekeyLimit` policy.
+    #[must_use]
+    pub fn with_rekey_limit(mut self, rekey_limit: RekeyLimit) -> Self {
+        self.rekey_limit = rekey_limit;
+        self
+    }
+
     /// Convert this configuration to a russh client Config.
     ///
     /// `inactivity_timeout` stays disabled for client sessions. A healthy
@@ -877,6 +892,7 @@ impl SshConnectionConfig {
         };
 
         Config {
+            limits: self.rekey_limit.to_russh_limits(),
             keepalive_interval: self.keepalive_interval.map(Duration::from_secs),
             keepalive_max: self.keepalive_max,
             inactivity_timeout: None,
diff --git a/src/ssh/tokio_client/connection_tests.rs b/src/ssh/tokio_client/connection_tests.rs
index 67f49b89..e9b64fab 100644
--- a/src/ssh/tokio_client/connection_tests.rs
+++ b/src/ssh/tokio_client/connection_tests.rs
@@ -33,7 +33,9 @@ use super::connection::{
     resolve_interface_address,
 };
 use super::proxy_command::{ProxyCommandConfig, ProxyMode};
-use crate::ssh::ssh_config::{IpQosPolicy, IpQosValue};
+use crate::ssh::ssh_config::{
+    IpQosPolicy, IpQosValue, RekeyDataLimit, RekeyLimit, RekeyTimeLimit,
+};
 use crate::ssh::{SessionPurpose, SshConfig};
 
 #[test]
@@ -889,3 +891,36 @@ fn ipv6_qos_sets_the_exact_traffic_class_byte() {
     apply_ip_qos(&socket, destination, IpQosValue::Class(0x48)).unwrap();
     assert_eq!(socket.tclass_v6().unwrap(), 0x48);
 }
+
+#[test]
+fn rekey_limit_reaches_the_russh_transport() {
+    let ssh_config = SshConfig::parse("Host target\n    RekeyLimit 4K 30s\n").unwrap();
+    let connection_config = SshConnectionConfigResolver::new()
+        .with_ssh_config(Some(ssh_config))
+        .resolve_for_host("target");
+
+    assert_eq!(
+        connection_config.rekey_limit,
+        RekeyLimit {
+            data: RekeyDataLimit::Bytes(4_096),
+            time: RekeyTimeLimit::Seconds(30),
+        }
+    );
+    let limits = connection_config.to_russh_config().limits;
+    assert_eq!(limits.rekey_write_limit, 4_096);
+    assert_eq!(limits.rekey_read_limit, 4_096);
+    assert_eq!(limits.rekey_time_limit, Duration::from_secs(30));
+}
+
+#[test]
+fn rekey_limit_default_none_preserves_russh_nonce_safety_ceiling() {
+    let ssh_config = SshConfig::parse("Host target\n    RekeyLimit default none\n").unwrap();
+    let connection_config = SshConnectionConfigResolver::new()
+        .with_ssh_config(Some(ssh_config))
+        .resolve_for_host("target");
+    let limits = connection_config.to_russh_config().limits;
+
+    assert_eq!(limits.rekey_write_limit, 1 << 30);
+    assert_eq!(limits.rekey_read_limit, 1 << 30);
+    assert_eq!(limits.rekey_time_limit, Duration::MAX);
+}
diff --git a/src/ssh/tokio_client/session.rs b/src/ssh/tokio_client/session.rs
index ce1adc95..bb369121 100644
--- a/src/ssh/tokio_client/session.rs
+++ b/src/ssh/tokio_client/session.rs
@@ -16,6 +16,7 @@
 
 use russh::client::Msg;
 use russh::{Channel, ChannelMsg};
+use tokio::io::{AsyncRead, AsyncReadExt};
 use tokio::sync::mpsc::Sender;
 
 use crate::ssh::{SessionPolicy, SessionRequest};
@@ -29,11 +30,35 @@ impl Client {
         policy: &SessionPolicy,
         sender: Sender<CommandOutput>,
     ) -> Result<u32, super::Error> {
+        self.execute_session_streaming_with_input(policy, sender, tokio::io::empty())
+            .await
+    }
+
+    /// Execute a resolved session while forwarding piped process input.
+    pub async fn execute_session_streaming_with_stdin(
+        &self,
+        policy: &SessionPolicy,
+        sender: Sender<CommandOutput>,
+    ) -> Result<u32, super::Error> {
+        self.execute_session_streaming_with_input(policy, sender, tokio::io::stdin())
+            .await
+    }
+
+    async fn execute_session_streaming_with_input<R>(
+        &self,
+        policy: &SessionPolicy,
+        sender: Sender<CommandOutput>,
+        input: R,
+    ) -> Result<u32, super::Error>
+    where
+        R: AsyncRead + Unpin,
+    {
         if matches!(policy.request, SessionRequest::None) {
             return Ok(0);
         }
         let channel = self.open_policy_channel(policy).await?;
-        self.drain_policy_channel(channel, sender).await
+        self.drain_policy_channel_with_input(channel, sender, input)
+            .await
     }
 
     pub async fn execute_session(
@@ -112,14 +137,41 @@ impl Client {
         Ok(channel)
     }
 
-    async fn drain_policy_channel(
+    async fn drain_policy_channel_with_input<R>(
         &self,
         mut channel: Channel<Msg>,
         sender: Sender<CommandOutput>,
-    ) -> Result<u32, super::Error> {
+        mut input: R,
+    ) -> Result<u32, super::Error>
+    where
+        R: AsyncRead + Unpin,
+    {
         let mut exit_status = None;
         let mut receiver_open = true;
-        while let Some(message) = channel.wait().await {
+        let mut input_open = true;
+        let mut input_buffer = [0_u8; 32 * 1024];
+        loop {
+            let message = if input_open {
+                tokio::select! {
+                    read = input.read(&mut input_buffer) => {
+                        match read.map_err(super::Error::IoError)? {
+                            0 => {
+                                channel.eof().await?;
+                                input_open = false;
+                            }
+                            count => channel.data(&input_buffer[..count]).await?,
+                        }
+                        continue;
+                    }
+                    message = channel.wait() => message,
+                }
+            } else {
+                channel.wait().await
+            };
+
+            let Some(message) = message else {
+                break;
+            };
             let output = match message {
                 ChannelMsg::Data { data } => Some(CommandOutput::StdOut(data)),
                 ChannelMsg::ExtendedData { data, ext: 1 } => Some(CommandOutput::StdErr(data)),

From 8afd28595734df15f5dc04081318a3ba34745588 Mon Sep 17 00:00:00 2001
From: Jeongkyu Shin <inureyes@gmail.com>
Date: Sun, 30 Aug 2026 13:07:00 +0900
Subject: [PATCH 2/5] fix(ssh): enforce inbound and idle rekey limits

The transport exposed read and time limits but only evaluated write bytes and checked elapsed time while flushing outbound work, leaving inbound-heavy and idle authenticated sessions outside the configured rekey policy.

Count decrypted packet payload bytes per key epoch, trigger authenticated client and server rekeys at the read threshold, reset counters when NEWKEYS completes, and schedule finite time limits as independent event-loop deadlines. Add focused counter, threshold, and idle tests plus a loopback transfer that observes a timer-driven rekey and completes repeated bidirectional byte-limit rekeys.

Refs #301
---
 crates/bssh-russh/src/cipher/mod.rs     |   2 +
 crates/bssh-russh/src/client/mod.rs     |  88 ++++++++++++++++-
 crates/bssh-russh/src/client/test.rs    | 122 ++++++++++++++++++++++++
 crates/bssh-russh/src/server/session.rs |  98 +++++++++++++++++++
 crates/bssh-russh/src/sshbuffer.rs      |  21 +++-
 5 files changed, 329 insertions(+), 2 deletions(-)

diff --git a/crates/bssh-russh/src/cipher/mod.rs b/crates/bssh-russh/src/cipher/mod.rs
index 508c2659..662d9974 100644
--- a/crates/bssh-russh/src/cipher/mod.rs
+++ b/crates/bssh-russh/src/cipher/mod.rs
@@ -345,10 +345,12 @@ pub(crate) async fn read<R: AsyncRead + Unpin>(
         .len()
         .checked_sub(padding_length)
         .ok_or(Error::IndexOutOfBounds)?;
+    let payload_len = plaintext_end.saturating_sub(PADDING_LENGTH_LEN);
 
     // Sequence numbers are on 32 bits and wrap.
     // https://tools.ietf.org/html/rfc4253#section-6.4
     buffer.seqn += Wrapping(1);
+    buffer.bytes = buffer.bytes.saturating_add(payload_len);
     buffer.len = 0;
 
     // Remove the padding
diff --git a/crates/bssh-russh/src/client/mod.rs b/crates/bssh-russh/src/client/mod.rs
index 6431237a..58ea2eb1 100644
--- a/crates/bssh-russh/src/client/mod.rs
+++ b/crates/bssh-russh/src/client/mod.rs
@@ -40,7 +40,6 @@ use std::convert::TryInto;
 use std::num::Wrapping;
 use std::pin::Pin;
 use std::sync::Arc;
-#[cfg(not(target_arch = "wasm32"))]
 use std::time::Duration;
 
 use futures::Future;
@@ -1315,6 +1314,9 @@ impl Session {
             // application output in its bounded receivers while a channel is
             // window-blocked.
             let can_receive_outbound = !self.kex.active() && !self.common.has_any_pending_data();
+            let rekey_timer =
+                crate::future_or_pending(self.rekey_time_remaining(), tokio::time::sleep);
+            pin!(rekey_timer);
             tokio::select! {
                 r = &mut reading => {
                     let (stream_read, mut buffer, mut opening_cipher) = match r {
@@ -1336,14 +1338,26 @@ impl Session {
                             result = self.process_disconnect(&pkt).map_err(H::Error::from);
                         } else {
                             self.common.received_data = true;
+                            let kex_was_active = self.kex.active();
                             reply(self, handler, kex_done_signal, &mut pkt).await?;
                             buffer.seqn = pkt.seqn; // TODO reply changes seqn internall, find cleaner way
+
+                            if kex_was_active && !self.kex.active() {
+                                buffer.bytes = 0;
+                            } else if self.read_rekey_limit_reached(buffer.bytes) {
+                                debug!("rekey limit reached after {} inbound bytes", buffer.bytes);
+                                self.initiate_rekey()?;
+                            }
                         }
                     }
 
                     std::mem::swap(&mut opening_cipher, &mut self.common.remote_to_local);
                     reading.set(start_reading(stream_read, buffer, opening_cipher));
                 }
+                () = &mut rekey_timer => {
+                    debug!("rekey time limit reached");
+                    self.initiate_rekey()?;
+                }
                 () = &mut keepalive_timer => {
                     if let Some(ref mut enc) = self.common.encrypted {
                         if matches!(enc.state, EncryptedState::Authenticated) {
@@ -1739,6 +1753,34 @@ impl Session {
         Ok(())
     }
 
+    fn read_rekey_limit_reached(&self, read_bytes: usize) -> bool {
+        !self.kex.active()
+            && self
+                .common
+                .encrypted
+                .as_ref()
+                .is_some_and(|enc| matches!(enc.state, EncryptedState::Authenticated))
+            && read_bytes >= self.common.config.limits.rekey_read_limit
+    }
+
+    fn rekey_time_remaining(&self) -> Option<Duration> {
+        if self.kex.active() {
+            return None;
+        }
+
+        let enc = self.common.encrypted.as_ref()?;
+        if !matches!(enc.state, EncryptedState::Authenticated) {
+            return None;
+        }
+
+        let limit = self.common.config.limits.rekey_time_limit;
+        if limit == Duration::MAX {
+            return None;
+        }
+
+        Some(limit.saturating_sub(Instant::now().duration_since(enc.last_rekey)))
+    }
+
     /// Flush the temporary cleartext buffer into the encryption
     /// buffer. This does *not* flush to the socket.
     fn flush(&mut self) -> Result<(), crate::Error> {
@@ -2022,6 +2064,50 @@ mod tests {
         );
     }
 
+    #[test]
+    fn automatic_rekey_starts_at_inbound_read_limit() {
+        let (mut session, sender, _replies) = keyboard_interactive_session();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(
+            1 << 30,
+            16,
+            std::time::Duration::from_secs(3600),
+        );
+        let encrypted = session.common.encrypted.as_mut().unwrap();
+        encrypted.state = EncryptedState::Authenticated;
+        encrypted.kex = KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        drop(sender);
+
+        assert!(!session.read_rekey_limit_reached(15));
+        assert!(session.read_rekey_limit_reached(16));
+        session.initiate_rekey().unwrap();
+
+        assert!(session.kex.active());
+    }
+
+    #[tokio::test]
+    async fn automatic_rekey_deadline_wakes_an_idle_authenticated_session() {
+        let (mut session, sender, _replies) = keyboard_interactive_session();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(1 << 30, 1 << 30, Duration::ZERO);
+        let encrypted = session.common.encrypted.as_mut().unwrap();
+        encrypted.state = EncryptedState::Authenticated;
+        encrypted.kex = KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        drop(sender);
+
+        let remaining = session
+            .rekey_time_remaining()
+            .expect("authenticated sessions with a finite limit need a deadline");
+        tokio::time::timeout(Duration::from_millis(50), tokio::time::sleep(remaining))
+            .await
+            .expect("an expired rekey deadline must wake without transport activity");
+        session.initiate_rekey().unwrap();
+
+        assert!(session.kex.active());
+    }
+
     #[cfg(feature = "flate2")]
     fn authenticated_session() -> Session {
         let config = Arc::new(Config::default());
diff --git a/crates/bssh-russh/src/client/test.rs b/crates/bssh-russh/src/client/test.rs
index 3aef6921..64860015 100644
--- a/crates/bssh-russh/src/client/test.rs
+++ b/crates/bssh-russh/src/client/test.rs
@@ -2,6 +2,7 @@
 mod tests {
     use std::collections::HashMap;
     use std::sync::{Arc, Mutex};
+    use std::sync::atomic::{AtomicUsize, Ordering};
 
     use log::debug;
     use ssh_key::PrivateKey;
@@ -83,6 +84,31 @@ mod tests {
         }
     }
 
+    struct CountingClient {
+        kex_count: Arc<AtomicUsize>,
+    }
+
+    impl Handler for CountingClient {
+        type Error = Error;
+
+        async fn check_server_key(
+            &mut self,
+            _: &PublicKeyOrCertificate,
+        ) -> Result<bool, Self::Error> {
+            Ok(true)
+        }
+
+        async fn kex_done(
+            &mut self,
+            _: Option<&[u8]>,
+            _: &crate::negotiation::Names,
+            _: &mut crate::client::Session,
+        ) -> Result<(), Self::Error> {
+            self.kex_count.fetch_add(1, Ordering::Relaxed);
+            Ok(())
+        }
+    }
+
     #[tokio::test]
     async fn test_client_connects_to_protocol_1_99() {
         let _ = env_logger::try_init();
@@ -164,4 +190,100 @@ mod tests {
             msg => panic!("Unexpected message {msg:?}"),
         }
     }
+
+    #[tokio::test]
+    async fn automatic_rekey_repeated_bidirectional_transfers_complete() {
+        let client_key = PrivateKey::random(&mut rng(), ssh_key::Algorithm::Ed25519).unwrap();
+
+        let mut server_config = server::Config::default();
+        server_config.auth_rejection_time = std::time::Duration::from_millis(1);
+        server_config.inactivity_timeout = None;
+        server_config.limits = crate::Limits::new(
+            8 * 1024,
+            8 * 1024,
+            std::time::Duration::from_secs(3600),
+        );
+        server_config
+            .keys
+            .push(PrivateKey::random(&mut rng(), ssh_key::Algorithm::Ed25519).unwrap());
+        let server_config = Arc::new(server_config);
+
+        let socket = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = socket.local_addr().unwrap();
+        let server = TestServer {
+            clients: Arc::new(Mutex::new(HashMap::new())),
+            id: 0,
+        };
+        let server_task = tokio::spawn(async move {
+            let (socket, _) = socket.accept().await.unwrap();
+            let running = server::run_stream(server_config, socket, server)
+                .await
+                .unwrap();
+            running.await
+        });
+
+        let mut client_config = Config::default();
+        client_config.limits = crate::Limits::new(
+            8 * 1024,
+            8 * 1024,
+            std::time::Duration::from_millis(50),
+        );
+        let kex_count = Arc::new(AtomicUsize::new(0));
+
+        tokio::time::timeout(std::time::Duration::from_secs(20), async {
+            let mut session = connect(
+                Arc::new(client_config),
+                addr,
+                CountingClient {
+                    kex_count: Arc::clone(&kex_count),
+                },
+            )
+            .await
+            .unwrap();
+            let authenticated = session
+                .authenticate_publickey(
+                    "rekey-user",
+                    PrivateKeyWithHashAlg::new(
+                        Arc::new(client_key),
+                        session.best_supported_rsa_hash().await.unwrap().flatten(),
+                    ),
+                )
+                .await
+                .unwrap();
+            assert!(authenticated.success());
+
+            while kex_count.load(Ordering::Relaxed) < 2 {
+                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+            }
+
+            let mut channel = session.channel_open_session().await.unwrap();
+            for round in 0_u8..6 {
+                let payload = vec![round; 32 * 1024];
+                channel.data(&payload[..]).await.unwrap();
+                let mut echoed = Vec::with_capacity(payload.len());
+                while echoed.len() < payload.len() {
+                    match channel.wait().await.unwrap() {
+                        crate::channels::ChannelMsg::Data { data } => {
+                            echoed.extend_from_slice(&data)
+                        }
+                        message => panic!("unexpected message during rekey transfer: {message:?}"),
+                    }
+                }
+                assert!(echoed == payload, "echo mismatch in rekey round {round}");
+            }
+
+            session
+                .disconnect(crate::Disconnect::ByApplication, "test complete", "")
+                .await
+                .unwrap();
+        })
+        .await
+        .expect("repeated bidirectional rekeys must not stall");
+
+        tokio::time::timeout(std::time::Duration::from_secs(2), server_task)
+            .await
+            .expect("server session must stop after client disconnect")
+            .unwrap()
+            .unwrap();
+    }
 }
diff --git a/crates/bssh-russh/src/server/session.rs b/crates/bssh-russh/src/server/session.rs
index 49237346..b98dafd3 100644
--- a/crates/bssh-russh/src/server/session.rs
+++ b/crates/bssh-russh/src/server/session.rs
@@ -710,6 +710,9 @@ impl Session {
                 }
             }
 
+            let rekey_timer =
+                future_or_pending(self.rekey_time_remaining(), tokio::time::sleep);
+            pin!(rekey_timer);
             tokio::select! {
                 r = &mut reading => {
                     let (stream_read, mut buffer, mut opening_cipher) = match r {
@@ -732,6 +735,7 @@ impl Session {
                         }
                         Some(_) => {
                             self.common.received_data = true;
+                            let kex_was_active = self.kex.active();
                             // TODO it'd be cleaner to just pass cipher to reply()
                             std::mem::swap(&mut opening_cipher, &mut self.common.remote_to_local);
 
@@ -742,10 +746,21 @@ impl Session {
                             buffer.seqn = pkt.seqn; // TODO reply changes seqn internall, find cleaner way
 
                             std::mem::swap(&mut opening_cipher, &mut self.common.remote_to_local);
+
+                            if kex_was_active && !self.kex.active() {
+                                buffer.bytes = 0;
+                            } else if self.read_rekey_limit_reached(buffer.bytes) {
+                                debug!("rekey limit reached after {} inbound bytes", buffer.bytes);
+                                self.initiate_rekey()?;
+                            }
                         }
                     }
                     reading.set(start_reading(stream_read, buffer, opening_cipher));
                 }
+                () = &mut rekey_timer => {
+                    debug!("rekey time limit reached");
+                    self.initiate_rekey()?;
+                }
                 () = &mut keepalive_timer => {
                     self.common.alive_timeouts = self.common.alive_timeouts.saturating_add(1);
                     if self.common.config.keepalive_max != 0 && self.common.alive_timeouts > self.common.config.keepalive_max {
@@ -891,6 +906,42 @@ impl Session {
         Ok(())
     }
 
+    fn initiate_rekey(&mut self) -> Result<(), Error> {
+        if let Some(ref mut enc) = self.common.encrypted {
+            enc.rekey_wanted = true;
+            self.flush()?;
+        }
+        Ok(())
+    }
+
+    fn read_rekey_limit_reached(&self, read_bytes: usize) -> bool {
+        !self.kex.active()
+            && self
+                .common
+                .encrypted
+                .as_ref()
+                .is_some_and(|enc| matches!(enc.state, EncryptedState::Authenticated))
+            && read_bytes >= self.common.config.limits.rekey_read_limit
+    }
+
+    fn rekey_time_remaining(&self) -> Option<std::time::Duration> {
+        if self.kex.active() {
+            return None;
+        }
+
+        let enc = self.common.encrypted.as_ref()?;
+        if !matches!(enc.state, EncryptedState::Authenticated) {
+            return None;
+        }
+
+        let limit = self.common.config.limits.rekey_time_limit;
+        if limit == std::time::Duration::MAX {
+            return None;
+        }
+
+        Some(limit.saturating_sub(Instant::now().duration_since(enc.last_rekey)))
+    }
+
     pub fn flush_pending(&mut self, channel: ChannelId) -> Result<usize, Error> {
         if let Some(ref mut enc) = self.common.encrypted {
             enc.flush_pending(channel)
@@ -1619,4 +1670,51 @@ mod tests {
             matches!(err, crate::Error::PacketSize(len) if len > crate::cipher::MAXIMUM_DECOMPRESSED_PACKET_LEN)
         );
     }
+
+    #[test]
+    fn automatic_rekey_server_starts_at_inbound_read_limit() {
+        let mut session = authenticated_session();
+        session.common.encrypted.as_mut().unwrap().kex =
+            KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(
+            1 << 30,
+            16,
+            std::time::Duration::from_secs(3600),
+        );
+
+        assert!(!session.read_rekey_limit_reached(15));
+        assert!(session.read_rekey_limit_reached(16));
+        session.initiate_rekey().unwrap();
+
+        assert!(session.kex.active());
+    }
+
+    #[tokio::test]
+    async fn automatic_rekey_server_deadline_wakes_an_idle_authenticated_session() {
+        let mut session = authenticated_session();
+        session.common.encrypted.as_mut().unwrap().kex =
+            KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(
+            1 << 30,
+            1 << 30,
+            std::time::Duration::ZERO,
+        );
+
+        let remaining = session
+            .rekey_time_remaining()
+            .expect("authenticated sessions with a finite limit need a deadline");
+        tokio::time::timeout(
+            std::time::Duration::from_millis(50),
+            tokio::time::sleep(remaining),
+        )
+        .await
+        .expect("an expired rekey deadline must wake without transport activity");
+        session.initiate_rekey().unwrap();
+
+        assert!(session.kex.active());
+    }
 }
diff --git a/crates/bssh-russh/src/sshbuffer.rs b/crates/bssh-russh/src/sshbuffer.rs
index 234693ab..4f2b993f 100644
--- a/crates/bssh-russh/src/sshbuffer.rs
+++ b/crates/bssh-russh/src/sshbuffer.rs
@@ -187,6 +187,25 @@ fn test_write_packet_matches_clear_cipher_write_output() {
     assert_eq!(writer.buffer().seqn, Wrapping(1));
 }
 
+#[tokio::test]
+async fn automatic_rekey_read_counter_tracks_payload_bytes() {
+    let payload = b"inbound rekey accounting";
+    let mut writer = PacketWriter::clear();
+    writer.packet_raw(payload).unwrap();
+    let encoded = writer.buffer().buffer.clone();
+    let (mut reader, mut sender) = tokio::io::duplex(encoded.len());
+    sender.write_all(&encoded).await.unwrap();
+    drop(sender);
+
+    let mut buffer = SSHBuffer::new();
+    let mut opening_key = cipher::clear::Key {};
+    cipher::read(&mut reader, &mut buffer, &mut opening_key)
+        .await
+        .unwrap();
+
+    assert_eq!(buffer.bytes, payload.len());
+}
+
 #[test]
 fn test_write_packet_restores_output_buffer_on_error() {
     let mut writer = PacketWriter::clear();
@@ -290,7 +309,7 @@ fn packet_bytes_compressed_matches_packet_output() {
 pub struct SSHBuffer {
     pub buffer: Vec<u8>,
     pub len: usize,   // next packet length.
-    pub bytes: usize, // total bytes written since the last rekey
+    pub bytes: usize, // total payload bytes read/written since the last rekey
     // Sequence numbers are on 32 bits and wrap.
     // https://tools.ietf.org/html/rfc4253#section-6.4
     pub seqn: Wrapping<u32>,

From 623b8504f970e676a017b8b3b92656044577394d Mon Sep 17 00:00:00 2001
From: Jeongkyu Shin <inureyes@gmail.com>
Date: Sun, 30 Aug 2026 13:23:25 +0900
Subject: [PATCH 3/5] fix(ssh): align automatic rekey epoch boundaries

Reset transport byte accounting whenever a new outbound cipher is installed, including the initial encrypted epoch. Gate server-initiated automatic rekeys on completed authentication and make the bidirectional transfer regression observe repeated byte-triggered exchanges.

Refs #301
---
 crates/bssh-russh/src/client/test.rs    | 25 ++++++++++++-------
 crates/bssh-russh/src/server/session.rs | 32 +++++++++++++++++++++++++
 crates/bssh-russh/src/sshbuffer.rs      | 17 +++++++++++++
 3 files changed, 66 insertions(+), 8 deletions(-)

diff --git a/crates/bssh-russh/src/client/test.rs b/crates/bssh-russh/src/client/test.rs
index 64860015..eaaf53b2 100644
--- a/crates/bssh-russh/src/client/test.rs
+++ b/crates/bssh-russh/src/client/test.rs
@@ -226,7 +226,7 @@ mod tests {
         client_config.limits = crate::Limits::new(
             8 * 1024,
             8 * 1024,
-            std::time::Duration::from_millis(50),
+            std::time::Duration::from_secs(3600),
         );
         let kex_count = Arc::new(AtomicUsize::new(0));
 
@@ -247,14 +247,11 @@ mod tests {
                         Arc::new(client_key),
                         session.best_supported_rsa_hash().await.unwrap().flatten(),
                     ),
-                )
-                .await
-                .unwrap();
+            )
+            .await
+            .unwrap();
             assert!(authenticated.success());
-
-            while kex_count.load(Ordering::Relaxed) < 2 {
-                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
-            }
+            let kex_count_before_transfer = kex_count.load(Ordering::Relaxed);
 
             let mut channel = session.channel_open_session().await.unwrap();
             for round in 0_u8..6 {
@@ -272,6 +269,18 @@ mod tests {
                 assert!(echoed == payload, "echo mismatch in rekey round {round}");
             }
 
+            tokio::time::timeout(std::time::Duration::from_secs(2), async {
+                while kex_count.load(Ordering::Relaxed) < kex_count_before_transfer + 2 {
+                    tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+                }
+            })
+            .await
+            .expect("byte thresholds must complete repeated rekeys during transfer");
+            assert!(
+                kex_count.load(Ordering::Relaxed) >= kex_count_before_transfer + 2,
+                "six threshold-crossing rounds must observe repeated key exchanges"
+            );
+
             session
                 .disconnect(crate::Disconnect::ByApplication, "test complete", "")
                 .await
diff --git a/crates/bssh-russh/src/server/session.rs b/crates/bssh-russh/src/server/session.rs
index b98dafd3..28de6ad5 100644
--- a/crates/bssh-russh/src/server/session.rs
+++ b/crates/bssh-russh/src/server/session.rs
@@ -896,6 +896,7 @@ impl Session {
                 &self.common.config.as_ref().limits,
                 &mut self.common.packet_writer,
             )? && self.kex == SessionKexState::Idle
+                && matches!(enc.state, EncryptedState::Authenticated)
             {
                 debug!("starting rekeying");
                 if enc.exchange.take().is_some() {
@@ -1691,6 +1692,37 @@ mod tests {
         assert!(session.kex.active());
     }
 
+    #[test]
+    fn automatic_rekey_server_waits_until_authentication_completes() {
+        let mut session = authenticated_session();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(
+            0,
+            0,
+            std::time::Duration::from_secs(3600),
+        );
+        let encrypted = session.common.encrypted.as_mut().unwrap();
+        encrypted.state = EncryptedState::WaitingAuthServiceRequest {
+            accepted: false,
+            sent: false,
+        };
+        encrypted.kex = KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+
+        session.flush().unwrap();
+        assert!(
+            !session.kex.active(),
+            "a pre-authentication server write limit must not start rekeying"
+        );
+
+        session.common.encrypted.as_mut().unwrap().state = EncryptedState::Authenticated;
+        session.flush().unwrap();
+        assert!(
+            session.kex.active(),
+            "the same server write limit must start rekeying after authentication"
+        );
+    }
+
     #[tokio::test]
     async fn automatic_rekey_server_deadline_wakes_an_idle_authenticated_session() {
         let mut session = authenticated_session();
diff --git a/crates/bssh-russh/src/sshbuffer.rs b/crates/bssh-russh/src/sshbuffer.rs
index 4f2b993f..fc9980b3 100644
--- a/crates/bssh-russh/src/sshbuffer.rs
+++ b/crates/bssh-russh/src/sshbuffer.rs
@@ -187,6 +187,19 @@ fn test_write_packet_matches_clear_cipher_write_output() {
     assert_eq!(writer.buffer().seqn, Wrapping(1));
 }
 
+#[test]
+fn installing_a_cipher_starts_a_new_write_key_epoch() {
+    let mut writer = PacketWriter::clear();
+    writer.packet_raw(b"initial key exchange").unwrap();
+    assert_ne!(writer.buffer().bytes, 0);
+
+    writer.set_cipher(Box::new(cipher::clear::Key {}));
+
+    assert_eq!(writer.buffer().bytes, 0);
+    writer.packet_raw(b"encrypted epoch").unwrap();
+    assert_eq!(writer.buffer().bytes, b"encrypted epoch".len());
+}
+
 #[tokio::test]
 async fn automatic_rekey_read_counter_tracks_payload_bytes() {
     let payload = b"inbound rekey accounting";
@@ -576,6 +589,10 @@ impl PacketWriter {
 
     pub fn set_cipher(&mut self, cipher: Box<dyn SealingKey + Send>) {
         self.cipher = cipher;
+        // Every cipher installation starts a new key epoch. In particular,
+        // discard the cleartext KEX packets counted before the initial keys
+        // were installed as well as the previous encrypted epoch on rekey.
+        self.write_buffer.bytes = 0;
     }
 
     pub fn reset_seqn(&mut self) {

From 4ca18a8cd6ee8b352bf4fc09212b24bd3f8508e2 Mon Sep 17 00:00:00 2001
From: Jeongkyu Shin <inureyes@gmail.com>
Date: Sun, 30 Aug 2026 17:14:32 +0900
Subject: [PATCH 4/5] fix(ssh): preserve per-hop rekey policy after integration

The #301 transplant now leaves all completed runtime keyword classifications active while retaining the delegated classification type for future issue waves without warning under the lint gate.

Extend the jump-chain regression with distinct bastion, destination-alias, and expanded-host RekeyLimit values so the final destination cannot silently re-resolve away from its original alias policy.

Refs #301
---
 src/jump/chain.rs                             | 19 +++++++++++++++++++
 .../ssh_config/parser/options/connection.rs   |  2 +-
 src/ssh/ssh_config/parser/options/support.rs  |  5 ++++-
 src/ssh/tokio_client/connection_tests.rs      |  4 +---
 4 files changed, 25 insertions(+), 5 deletions(-)

diff --git a/src/jump/chain.rs b/src/jump/chain.rs
index 3c9eab52..30183cde 100644
--- a/src/jump/chain.rs
+++ b/src/jump/chain.rs
@@ -510,6 +510,7 @@ fn intermediate_jump_hop_context(jump_host: &JumpHost, hop: usize) -> String {
 mod tests {
     use super::*;
     use crate::ssh::SshConfig;
+    use crate::ssh::ssh_config::{RekeyDataLimit, RekeyLimit, RekeyTimeLimit};
     use crate::ssh::tokio_client::AddressFamily;
 
     #[test]
@@ -573,6 +574,7 @@ Host bastion
     Compression yes
     ServerAliveInterval 11
     ServerAliveCountMax 2
+    RekeyLimit 4K 30s
 
 Host target-alias
     HostName effective-target
@@ -580,12 +582,14 @@ Host target-alias
     Compression no
     ServerAliveInterval 22
     ServerAliveCountMax 4
+    RekeyLimit 8K 60s
 
 Host effective-target
     AddressFamily inet
     Compression yes
     ServerAliveInterval 99
     ServerAliveCountMax 9
+    RekeyLimit 16K 90s
 "#,
         )
         .expect("valid ssh_config");
@@ -601,11 +605,26 @@ Host effective-target
         assert!(bastion.compression);
         assert_eq!(bastion.keepalive_interval, Some(11));
         assert_eq!(bastion.keepalive_max, 2);
+        assert_eq!(
+            bastion.rekey_limit,
+            RekeyLimit {
+                data: RekeyDataLimit::Bytes(4 * 1024),
+                time: RekeyTimeLimit::Seconds(30),
+            }
+        );
 
         let target = chain.destination_connection_config();
         assert_eq!(target.address_family, AddressFamily::V6);
         assert!(!target.compression);
         assert_eq!(target.keepalive_interval, Some(22));
         assert_eq!(target.keepalive_max, 4);
+        assert_eq!(
+            target.rekey_limit,
+            RekeyLimit {
+                data: RekeyDataLimit::Bytes(8 * 1024),
+                time: RekeyTimeLimit::Seconds(60),
+            },
+            "the destination must retain its original alias policy instead of re-resolving HostName"
+        );
     }
 }
diff --git a/src/ssh/ssh_config/parser/options/connection.rs b/src/ssh/ssh_config/parser/options/connection.rs
index 5e839142..5e11335d 100644
--- a/src/ssh/ssh_config/parser/options/connection.rs
+++ b/src/ssh/ssh_config/parser/options/connection.rs
@@ -17,9 +17,9 @@
 //! Handles connection-related configuration options including keepalive
 //! settings, timeouts, compression, and network settings.
 
-use crate::ssh::ssh_config::{IpQosPolicy, RekeyLimit};
 use crate::ssh::ssh_config::parser::helpers::parse_yes_no;
 use crate::ssh::ssh_config::types::SshHostConfig;
+use crate::ssh::ssh_config::{IpQosPolicy, RekeyLimit};
 use anyhow::{Context, Result};
 
 /// Parse connection-related SSH configuration options
diff --git a/src/ssh/ssh_config/parser/options/support.rs b/src/ssh/ssh_config/parser/options/support.rs
index 3ad6a4ed..25236732 100644
--- a/src/ssh/ssh_config/parser/options/support.rs
+++ b/src/ssh/ssh_config/parser/options/support.rs
@@ -4,6 +4,9 @@
 pub(super) enum KeywordSupport {
     Runtime(RuntimeConsumer),
     Unimplemented,
+    // Keep the classification available for future split issue waves even
+    // when the current wave has no remaining delegated keywords.
+    #[allow(dead_code)]
     Delegated(u32),
 }
 
@@ -24,7 +27,7 @@ pub(super) struct KeywordSpec {
     pub support: KeywordSupport,
 }
 
-use KeywordSupport::{Delegated, Runtime, Unimplemented};
+use KeywordSupport::{Runtime, Unimplemented};
 use RuntimeConsumer::{
     Authentication, Forwarding, HostVerification, NodeResolution, Proxy, Session, Transport,
 };
diff --git a/src/ssh/tokio_client/connection_tests.rs b/src/ssh/tokio_client/connection_tests.rs
index e9b64fab..536baef7 100644
--- a/src/ssh/tokio_client/connection_tests.rs
+++ b/src/ssh/tokio_client/connection_tests.rs
@@ -33,9 +33,7 @@ use super::connection::{
     resolve_interface_address,
 };
 use super::proxy_command::{ProxyCommandConfig, ProxyMode};
-use crate::ssh::ssh_config::{
-    IpQosPolicy, IpQosValue, RekeyDataLimit, RekeyLimit, RekeyTimeLimit,
-};
+use crate::ssh::ssh_config::{IpQosPolicy, IpQosValue, RekeyDataLimit, RekeyLimit, RekeyTimeLimit};
 use crate::ssh::{SessionPurpose, SshConfig};
 
 #[test]

From 3794d6c4a5e9c10fb5895fc01df8d7d46cbe5b4c Mon Sep 17 00:00:00 2001
From: Jeongkyu Shin <inureyes@gmail.com>
Date: Sun, 30 Aug 2026 17:40:13 +0900
Subject: [PATCH 5/5] fix(ssh): allow rekey immediately after authentication

---
 crates/bssh-russh/src/server/session.rs | 118 ++++++++++++++++++++++--
 1 file changed, 111 insertions(+), 7 deletions(-)

diff --git a/crates/bssh-russh/src/server/session.rs b/crates/bssh-russh/src/server/session.rs
index 28de6ad5..32261b55 100644
--- a/crates/bssh-russh/src/server/session.rs
+++ b/crates/bssh-russh/src/server/session.rs
@@ -891,22 +891,45 @@ impl Session {
 
     /// Flush the session, i.e. encrypt the pending buffer.
     pub fn flush(&mut self) -> Result<(), Error> {
+        let mut start_rekey = false;
         if let Some(ref mut enc) = self.common.encrypted {
             if enc.flush(
                 &self.common.config.as_ref().limits,
                 &mut self.common.packet_writer,
             )? && self.kex == SessionKexState::Idle
-                && matches!(enc.state, EncryptedState::Authenticated)
+                && Self::automatic_rekey_eligible(&enc.state)
             {
                 debug!("starting rekeying");
                 if enc.exchange.take().is_some() {
-                    self.begin_rekey()?;
+                    // USERAUTH_SUCCESS was just flushed with the pre-authentication
+                    // compression state. If automatic rekeying is the first
+                    // post-authentication traffic, activate delayed server
+                    // compression before emitting KEXINIT, just as the next peer
+                    // packet would normally do.
+                    if matches!(enc.state, EncryptedState::InitCompression) {
+                        if enc.server_compression.is_deferred() {
+                            enc.server_compression
+                                .init_compress(self.common.packet_writer.compress());
+                        }
+                        enc.state = EncryptedState::Authenticated;
+                    }
+                    start_rekey = true;
                 }
             }
         }
+        if start_rekey {
+            self.begin_rekey()?;
+        }
         Ok(())
     }
 
+    fn automatic_rekey_eligible(state: &EncryptedState) -> bool {
+        matches!(
+            state,
+            EncryptedState::InitCompression | EncryptedState::Authenticated
+        )
+    }
+
     fn initiate_rekey(&mut self) -> Result<(), Error> {
         if let Some(ref mut enc) = self.common.encrypted {
             enc.rekey_wanted = true;
@@ -921,7 +944,7 @@ impl Session {
                 .common
                 .encrypted
                 .as_ref()
-                .is_some_and(|enc| matches!(enc.state, EncryptedState::Authenticated))
+                .is_some_and(|enc| Self::automatic_rekey_eligible(&enc.state))
             && read_bytes >= self.common.config.limits.rekey_read_limit
     }
 
@@ -931,7 +954,7 @@ impl Session {
         }
 
         let enc = self.common.encrypted.as_ref()?;
-        if !matches!(enc.state, EncryptedState::Authenticated) {
+        if !Self::automatic_rekey_eligible(&enc.state) {
             return None;
         }
 
@@ -1556,8 +1579,11 @@ mod tests {
     use std::num::Wrapping;
     use std::sync::Arc;
 
+    use ssh_encoding::Encode;
+
     use super::*;
-    use crate::compression::{Compression, Decompress};
+    use crate::auth::{AuthRequest, MethodSet};
+    use crate::compression::{Compress, Compression, Decompress};
     use crate::kex::{KEXES, NONE, SessionKexState};
     use crate::session::{CommonSession, Encrypted, EncryptedState, Exchange};
     use crate::sshbuffer::{IncomingSshPacket, PacketWriter, SSHBuffer};
@@ -1569,6 +1595,24 @@ mod tests {
         type Error = crate::Error;
     }
 
+    struct AcceptNoneHandler;
+
+    impl crate::server::Handler for AcceptNoneHandler {
+        type Error = crate::Error;
+
+        async fn auth_none(&mut self, _user: &str) -> Result<Auth, Self::Error> {
+            Ok(Auth::Accept)
+        }
+    }
+
+    fn none_auth_request(user: &str) -> Vec<u8> {
+        let mut packet = vec![crate::msg::USERAUTH_REQUEST];
+        user.encode(&mut packet).unwrap();
+        "ssh-connection".encode(&mut packet).unwrap();
+        "none".encode(&mut packet).unwrap();
+        packet
+    }
+
     fn authenticated_session() -> Session {
         let config = Arc::new(crate::server::Config::default());
         let (priority_sender, priority_receiver) = tokio::sync::mpsc::unbounded_channel();
@@ -1708,6 +1752,7 @@ mod tests {
             sent: false,
         };
         encrypted.kex = KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        encrypted.server_compression = Compression::ZlibOpenSSH;
 
         session.flush().unwrap();
         assert!(
@@ -1715,12 +1760,71 @@ mod tests {
             "a pre-authentication server write limit must not start rekeying"
         );
 
-        session.common.encrypted.as_mut().unwrap().state = EncryptedState::Authenticated;
+        session.common.encrypted.as_mut().unwrap().state = EncryptedState::InitCompression;
         session.flush().unwrap();
         assert!(
             session.kex.active(),
-            "the same server write limit must start rekeying after authentication"
+            "the same server write limit must start rekeying after auth succeeds"
         );
+        assert!(
+            matches!(
+                session.common.encrypted.as_ref().unwrap().state,
+                EncryptedState::Authenticated
+            ),
+            "automatic rekey must complete the delayed-compression state transition"
+        );
+        assert!(
+            matches!(session.common.packet_writer.compress(), Compress::Zlib(_)),
+            "KEXINIT after auth success must use delayed server compression"
+        );
+    }
+
+    #[tokio::test]
+    async fn automatic_rekey_server_auth_success_arms_idle_deadline_before_next_packet() {
+        let mut session = authenticated_session();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(1 << 30, 1 << 30, std::time::Duration::ZERO);
+        let encrypted = session.common.encrypted.as_mut().unwrap();
+        encrypted.state =
+            EncryptedState::WaitingAuthRequest(AuthRequest::server(MethodSet::server_supported()));
+        encrypted.kex = KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        encrypted.server_compression = Compression::ZlibOpenSSH;
+
+        session
+            .process_packet(&mut AcceptNoneHandler, &none_auth_request("idle-user"))
+            .await
+            .unwrap();
+
+        assert!(matches!(
+            session.common.encrypted.as_ref().unwrap().state,
+            EncryptedState::InitCompression
+        ));
+        assert!(matches!(
+            session.common.packet_writer.compress(),
+            Compress::None
+        ));
+        let remaining = session
+            .rekey_time_remaining()
+            .expect("auth success must arm the finite rekey deadline before another peer packet");
+        tokio::time::timeout(
+            std::time::Duration::from_millis(50),
+            tokio::time::sleep(remaining),
+        )
+        .await
+        .expect("the idle post-authentication rekey deadline must wake");
+
+        session.initiate_rekey().unwrap();
+
+        assert!(session.kex.active());
+        assert!(matches!(
+            session.common.encrypted.as_ref().unwrap().state,
+            EncryptedState::Authenticated
+        ));
+        assert!(matches!(
+            session.common.packet_writer.compress(),
+            Compress::Zlib(_)
+        ));
     }
 
     #[tokio::test]