Rekey sending, avoid running out of send buffer

fuzz/Cargo.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-honggfuzz = { version = "=0.5.58", optional = true }
+honggfuzz = { version = "=0.5.60", optional = true }
 afl = { version = "*", optional = true }
 sunset = { workspace = true, features = ["arbitrary"] }
 sunset-sshwire-derive.workspace = true

fuzz/src/common.rs

@@ -114,8 +114,15 @@ where
 
 #[allow(unused)]
 fn check_error(r: Result<()>) {
+    use packets::MessageNumber::*;
     if let Err(e) = r {
         match e {
+            Error::BusySend { ref packet, unsupported: true } => match packet {
+                // Don't fail for userauth or service accept. In real operation
+                // they would happen early after KEX when there is space.
+                SSH_MSG_USERAUTH_SUCCESS | SSH_MSG_SERVICE_ACCEPT => (),
+                _ => panic!("Unexpected packet type for {e:#?}"),
+            },
             // Errors that should not occur.
             // May indicate a bug in this fuzz harness.
             Error::BadChannel { .. }

src/channel.rs

@@ -200,8 +200,7 @@ impl Channels {
         Ok(p)
     }
 
-    /// Informs the channel layer that an incoming packet has been read out,
-    /// so a window adjustment can be sent.
+    /// Informs the channel layer that an incoming packet has been read out.
     pub(crate) fn finished_read(
         &mut self,
         num: ChanNum,
@@ -210,9 +209,7 @@ impl Channels {
     ) -> Result<()> {
         let ch = self.get_mut(num)?;
         ch.finished_input(len);
-        if let Some(w) = ch.check_window_adjust()? {
-            s.send(w)?;
-        }
+        ch.check_send_window_adjust(s);
         Ok(())
     }
 
@@ -232,6 +229,27 @@ impl Channels {
         self.get(num).is_ok_and(|c| c.valid_send(dt))
     }
 
+    pub fn progress(&mut self, s: &mut TrafSend) -> DispatchEvent {
+        for ch in self.ch.iter_mut().filter_map(|c| c.as_mut()) {
+            ch.check_send_window_adjust(s);
+
+            if ch.open_confirmed {
+                ch.open_confirmed = false;
+                match ch.ty {
+                    ChanType::Session => {
+                        return DispatchEvent::CliEvent(CliEventId::SessionOpened(
+                            ch.num(),
+                        ));
+                    }
+                    ChanType::Tcp => {
+                        trace!("TODO tcp channel")
+                    }
+                }
+            }
+        }
+        DispatchEvent::None
+    }
+
     /// Wake the channel with a ready input data packet.
     pub fn wake_read(&mut self, num: ChanNum, dt: ChanData, is_client: bool) {
         if let Ok(ch) = self.get_mut(num) {
@@ -398,17 +416,8 @@ impl Channels {
                             window: p.initial_window as usize,
                         });
 
-                        match ch.ty {
-                            ChanType::Session => {
-                                ev = DispatchEvent::CliEvent(
-                                    CliEventId::SessionOpened(ch.num()),
-                                );
-                            }
-                            ChanType::Tcp => {
-                                trace!("TODO tcp channel")
-                            }
-                        }
-
+                        // A future progress() will notify the application.
+                        ch.open_confirmed = true;
                         ch.state = ChanState::Normal;
                     }
                     _ => {
@@ -743,6 +752,12 @@ pub(crate) struct Channel {
 
     full_window: usize,
 
+    /// Set when Open Confirmation is received.
+    ///
+    /// A subsequent `progress()` will emit a `SessionOpened` event
+    /// for the application to handle.
+    open_confirmed: bool,
+
     /// Set once application has called `done()`. The channel
     /// will only be removed from the list
     /// (allowing channel number re-use) if `app_done` is set
@@ -772,6 +787,7 @@ impl Channel {
             send: None,
             pending_adjust: 0,
             full_window: config::DEFAULT_WINDOW,
+            open_confirmed: false,
             app_done: false,
             read_waker: None,
             write_waker: None,
@@ -1030,16 +1046,22 @@ impl Channel {
         true
     }
 
-    /// Returns a window adjustment packet if required
-    fn check_window_adjust(&mut self) -> Result<Option<Packet<'_>>> {
-        let num = self.send.as_mut().trap()?.num;
+    /// Send a window adjust packet if required.
+    fn check_send_window_adjust(&mut self, s: &mut TrafSend) {
         if self.pending_adjust > self.full_window / 2 {
             let adjust = self.pending_adjust as u32;
-            self.pending_adjust = 0;
-            let p = packets::ChannelWindowAdjust { num, adjust }.into();
-            Ok(Some(p))
-        } else {
-            Ok(None)
+            let Some(sdir) = self.send.as_mut() else {
+                return;
+            };
+            let num = sdir.num;
+            let p = packets::ChannelWindowAdjust { num, adjust };
+            match s.send(p) {
+                Ok(()) => self.pending_adjust = 0,
+                Err(Error::BusySend { .. }) => {
+                    // Do nothing, the adjustment will be sent later.
+                }
+                Err(e) => debug_assert!(false, "Window adjust send failed {e:?}"),
+            }
         }
     }
 }

src/conn.rs

@@ -32,6 +32,12 @@ pub(crate) struct Conn<CS: CliServ> {
 
     cliserv: CS,
 
+    /// Algorithm preferences for KEX
+    ///
+    /// This must remain unmodified during a key exchange.
+    /// The same config will be serialised both for sending
+    /// and receiving kexinit, and possibly also for DelayedPacket
+    /// sending.
     algo_conf: AlgoConfig,
 
     parse_ctx: ParseContext,
@@ -263,18 +269,27 @@ impl<CS: CliServ> Conn<CS> {
     }
 
     /// Updates `ConnState` and sends any packets required to progress the connection state.
-    // TODO can this just move to the bottom of handle_payload(), and make module-private?
     pub(crate) fn progress(
         &mut self,
         s: &mut TrafSend,
     ) -> Result<Dispatched, Error> {
+        self.kex.progress(&self.algo_conf, s)?;
+
+        if !self.is_kex_sending() {
+            let event = self.channels.progress(s);
+            if !event.is_none() {
+                // TODO better Dispatched constructor
+                return Ok(Dispatched { event, disconnect: false });
+            }
+        }
+
         let mut disp = Dispatched::default();
         match self.state {
             ConnState::SendIdent => {
                 s.send_version()?;
                 // send early to avoid round trip latency
                 // TODO: first_follows would have a second packet here
-                self.kex.send_kexinit(&self.algo_conf, s)?;
+                self.kex.start_kexinit(s);
                 disp.event = DispatchEvent::Progressed;
                 self.state = ConnState::ReceiveIdent
             }
@@ -292,12 +307,8 @@ impl<CS: CliServ> Conn<CS> {
                 }
             }
             ConnState::PreAuth => {
-                // TODO. need to figure how we'll do "unbounded" responses
-                // and backpressure. can_output() should have a size check?
-                if s.can_output() {
-                    if let Some(cli) = self.try_mut_client() {
-                        disp.event = cli.auth.progress();
-                    }
+                if let Some(cli) = self.try_mut_client() {
+                    disp.event = cli.auth.progress();
                 }
                 // send userauth request
             }
@@ -307,8 +318,6 @@ impl<CS: CliServ> Conn<CS> {
         }
         trace!("-> {:?}, {disp:?}", self.state);
 
-        // TODO: if keys.seq > MAX_REKEY then we must rekey for security.
-
         Ok(disp)
     }
 
@@ -366,7 +375,7 @@ impl<CS: CliServ> Conn<CS> {
                     error::SSHProto.fail()
                 }
             }
-        } else if !matches!(self.kex, Kex::Idle | Kex::KexInit { .. }) {
+        } else if self.kex.is_receiving() {
             // Normal KEX only allows certain packets
             match p.category() {
                 packets::Category::All => Ok(()),
@@ -403,8 +412,9 @@ impl<CS: CliServ> Conn<CS> {
         self.sess_id.is_none()
     }
 
-    pub fn kex_is_idle(&self) -> bool {
-        matches!(self.kex, Kex::Idle)
+    /// True if KexInit has not been sent.
+    pub fn is_kex_sending(&self) -> bool {
+        self.kex.is_sending()
     }
 
     pub fn dispatch_packet(
@@ -590,8 +600,6 @@ impl Conn<Client> {
         &self,
         payload: &'f [u8],
     ) -> Result<PubKey<'f>> {
-        self.client()?;
-
         let packet = self.packet(payload)?;
         if let Packet::KexDHReply(p) = packet {
             Ok(p.k_s.0)
@@ -604,7 +612,6 @@ impl Conn<Client> {
         &mut self,
         payload: &'p [u8],
     ) -> Result<CliSessionExit<'p>> {
-        self.client()?;
         let packet = self.packet(payload)?;
         CliSessionExit::new(&packet)
     }
@@ -613,7 +620,6 @@ impl Conn<Client> {
         &mut self,
         payload: &'p [u8],
     ) -> Result<Banner<'p>> {
-        self.client()?;
         if let Packet::UserauthBanner(b) = self.packet(payload)? {
             Ok(Banner(b))
         } else {
@@ -629,8 +635,6 @@ impl Conn<Server> {
         s: &mut TrafSend,
         keys: &[&SignKey],
     ) -> Result<()> {
-        self.server()?;
-
         let packet = self.packet(payload)?;
         if let Packet::KexDHInit(p) = packet {
             self.kex.resume_kexdhinit(
@@ -649,8 +653,6 @@ impl Conn<Server> {
         &self,
         payload: &'f [u8],
     ) -> Result<TextString<'f>> {
-        self.server()?;
-
         let packet = self.packet(payload)?;
         if let Packet::UserauthRequest(UserauthRequest {
             method: AuthMethod::Password(m),
@@ -666,8 +668,6 @@ impl Conn<Server> {
         &self,
         payload: &'f [u8],
     ) -> Result<PubKey<'f>> {
-        self.server()?;
-
         let packet = self.packet(payload)?;
         if let Packet::UserauthRequest(UserauthRequest {
             method: AuthMethod::PubKey(m),