Client Timeout

russh/src/client/mod.rs

@@ -103,7 +103,9 @@ use crate::key::PubKey;
 use crate::session::{CommonSession, EncryptedState, Exchange, Kex, KexDhDone, KexInit, NewKeys};
 use crate::ssh_read::SshRead;
 use crate::sshbuffer::{SSHBuffer, SshId};
-use crate::{auth, msg, negotiation, ChannelId, ChannelOpenFailure, Disconnect, Limits, Sig};
+use crate::{
+    auth, msg, negotiation, timeout, ChannelId, ChannelOpenFailure, Disconnect, Limits, Sig,
+};
 
 mod encrypted;
 mod kex;
@@ -772,6 +774,8 @@ impl Session {
         pin!(reading);
         pin!(time_for_keepalive);
 
+        let delay = self.common.config.inactivity_timeout;
+
         #[allow(clippy::panic)] // false positive in select! macro
         while !self.common.disconnected {
             tokio::select! {
@@ -852,6 +856,10 @@ impl Session {
                         }
                     }
                 }
+                _ = timeout(delay) => {
+                    debug!("timeout");
+                    break
+                },
             }
             self.flush()?;
             if !self.common.write_buffer.buffer.is_empty() {
@@ -874,9 +882,10 @@ impl Session {
             }
         }
         debug!("disconnected");
-        if self.common.disconnected {
-            stream_write.shutdown().await.map_err(crate::Error::from)?;
-        }
+        self.receiver.close();
+        self.inbound_channel_receiver.close();
+        stream_write.shutdown().await.map_err(crate::Error::from)?;
+
         Ok(())
     }
 

russh/src/lib.rs

@@ -465,3 +465,11 @@ impl ChannelParams {
         self.confirmed = true;
     }
 }
+
+pub(crate) async fn timeout(delay: Option<std::time::Duration>) {
+    if let Some(delay) = delay {
+        tokio::time::sleep(delay).await
+    } else {
+        futures::future::pending().await
+    };
+}

russh/src/server/encrypted.rs

@@ -46,7 +46,7 @@ impl Session {
         // Either this packet is a KEXINIT, in which case we start a key re-exchange.
 
         #[allow(clippy::unwrap_used)]
-        let mut enc = self.common.encrypted.as_mut().unwrap();
+        let enc = self.common.encrypted.as_mut().unwrap();
         if buf.first() == Some(&msg::KEXINIT) {
             debug!("Received rekeying request");
             // If we're not currently rekeying, but `buf` is a rekey request
@@ -142,7 +142,7 @@ impl Session {
         };
 
         #[allow(clippy::unwrap_used)]
-        let mut enc = self.common.encrypted.as_mut().unwrap();
+        let enc = self.common.encrypted.as_mut().unwrap();
         // If we've successfully read a packet.
         match enc.state {
             EncryptedState::WaitingAuthServiceRequest {

russh/src/server/mod.rs

@@ -650,14 +650,6 @@ thread_local! {
     static B2: RefCell<CryptoVec> = RefCell::new(CryptoVec::new());
 }
 
-pub(crate) async fn timeout(delay: Option<std::time::Duration>) {
-    if let Some(delay) = delay {
-        tokio::time::sleep(delay).await
-    } else {
-        futures::future::pending().await
-    };
-}
-
 async fn start_reading<R: AsyncRead + Unpin>(
     mut stream_read: R,
     mut buffer: SSHBuffer,