CVE-2026-48110

@@ -27,7 +27,7 @@ use crate::cert::PublicKeyOrCertificate;
 use crate::client::{Handler, Msg, Prompt, Reply, Session};
 use crate::helpers::{AlgorithmExt, EncodedExt, NameList, sign_with_hash_alg};
 use crate::keys::key::parse_public_key;
-use crate::parsing::{ChannelOpenConfirmation, ChannelType, OpenChannelMessage};
+use crate::parsing::{ChannelOpenConfirmation, ChannelType, OpenChannelMessage, ensure_end};
 use crate::session::{Encrypted, EncryptedState, GlobalRequestResponse};
 use crate::{
     Channel, ChannelId, ChannelMsg, ChannelOpenFailure, ChannelParams, Error, MethodSet, Sig, auth,
@@ -72,6 +72,7 @@ impl Session {
                     match buf.split_first() {
                         Some((&msg::SERVICE_ACCEPT, mut r)) => {
                             if map_err!(Bytes::decode(&mut r))?.as_ref() == b"ssh-userauth" {
+                                map_err!(ensure_end(&r))?;
                                 *accepted = true;
                                 if let Some(ref meth) = self.common.auth_method {
                                     let len = enc.write.len();
@@ -98,7 +99,8 @@ impl Session {
                 EncryptedState::WaitingAuthRequest(ref mut auth_request) => {
                     trace!("waiting auth request, {:?}", buf.first(),);
                     match buf.split_first() {
-                        Some((&msg::USERAUTH_SUCCESS, _)) => {
+                        Some((&msg::USERAUTH_SUCCESS, r)) => {
+                            map_err!(ensure_end(&r))?;
                             debug!("userauth_success");
                             self.sender
                                 .send(Reply::AuthSuccess)
@@ -111,6 +113,8 @@ impl Session {
                         }
                         Some((&msg::USERAUTH_BANNER, mut r)) => {
                             let banner = map_err!(String::decode(&mut r))?;
+                            let _language_tag = map_err!(String::decode(&mut r))?;
+                            map_err!(ensure_end(&r))?;
                             client.auth_banner(&banner, self).await?;
                             return Ok(());
                         }
@@ -120,6 +124,7 @@ impl Session {
                             let remaining_methods: MethodSet =
                                 (&map_err!(NameList::decode(&mut r))?).into();
                             let partial_success = map_err!(u8::decode(&mut r))? != 0;
+                            map_err!(ensure_end(&r))?;
                             debug!(
                                 "remaining methods {remaining_methods:?}, partial success {partial_success:?}"
                             );
@@ -146,6 +151,9 @@ impl Session {
                             }) = auth_request.current
                             {
                                 debug!("userauth_pk_ok");
+                                let _algo = map_err!(String::decode(&mut r))?;
+                                let _key = map_err!(Bytes::decode(&mut r))?;
+                                map_err!(ensure_end(&r))?;
                                 *sent_pk_ok = true;
                             } else if let Some(auth::CurrentRequest::KeyboardInteractive {
                                 ..
@@ -173,6 +181,7 @@ impl Session {
                                         echo,
                                     });
                                 }
+                                map_err!(ensure_end(&r))?;
 
                                 // send challenges to caller
                                 self.sender
@@ -324,6 +333,7 @@ impl Session {
                 }
             }
         }
+        ensure_end(r)?;
         Ok(())
     }
 
@@ -351,6 +361,7 @@ impl Session {
             Some((&msg::CHANNEL_OPEN_CONFIRMATION, mut reader)) => {
                 debug!("channel_open_confirmation");
                 let msg = map_err!(ChannelOpenConfirmation::decode(&mut reader))?;
+                map_err!(ensure_end(&reader))?;
                 let local_id = ChannelId(msg.recipient_channel);
 
                 if let Some(ref mut enc) = self.common.encrypted {
@@ -389,6 +400,7 @@ impl Session {
             Some((&msg::CHANNEL_CLOSE, mut r)) => {
                 debug!("channel_close");
                 let channel_num = map_err!(ChannelId::decode(&mut r))?;
+                map_err!(ensure_end(&r))?;
                 if let Some(ref mut enc) = self.common.encrypted {
                     // The CHANNEL_CLOSE message must be sent to the server at this point or the session
                     // will not be released.
@@ -406,6 +418,7 @@ impl Session {
             Some((&msg::CHANNEL_EOF, mut r)) => {
                 debug!("channel_eof");
                 let channel_num = map_err!(ChannelId::decode(&mut r))?;
+                map_err!(ensure_end(&r))?;
                 if let Some(chan) = self.channels.get(&channel_num) {
                     let _ = chan.send(ChannelMsg::Eof).await;
                 }
@@ -418,6 +431,7 @@ impl Session {
                     .unwrap_or(ChannelOpenFailure::Unknown);
                 let descr = map_err!(String::decode(&mut r))?;
                 let language = map_err!(String::decode(&mut r))?;
+                map_err!(ensure_end(&r))?;
                 if let Some(ref mut enc) = self.common.encrypted {
                     enc.channels.remove(&channel_num);
                 }
@@ -436,6 +450,7 @@ impl Session {
                 trace!("channel_data");
                 let channel_num = map_err!(ChannelId::decode(&mut r))?;
                 let data = map_err!(Bytes::decode(&mut r))?;
+                map_err!(ensure_end(&r))?;
                 let target = self.common.config.window_size;
                 if let Some(ref mut enc) = self.common.encrypted {
                     if enc.adjust_window_size(channel_num, &data, target)? {
@@ -458,6 +473,7 @@ impl Session {
                 let channel_num = map_err!(ChannelId::decode(&mut r))?;
                 let extended_code = map_err!(u32::decode(&mut r))?;
                 let data = map_err!(Bytes::decode(&mut r))?;
+                map_err!(ensure_end(&r))?;
                 let target = self.common.config.window_size;
                 if let Some(ref mut enc) = self.common.encrypted {
                     if enc.adjust_window_size(channel_num, &data, target)? {
@@ -490,6 +506,7 @@ impl Session {
                     "xon-xoff" => {
                         map_err!(u8::decode(&mut r))?; // should be 0.
                         let client_can_do = map_err!(u8::decode(&mut r))? != 0;
+                        map_err!(ensure_end(&r))?;
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan.send(ChannelMsg::XonXoff { client_can_do }).await;
                         }
@@ -498,6 +515,7 @@ impl Session {
                     "exit-status" => {
                         map_err!(u8::decode(&mut r))?; // should be 0.
                         let exit_status = map_err!(u32::decode(&mut r))?;
+                        map_err!(ensure_end(&r))?;
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan.send(ChannelMsg::ExitStatus { exit_status }).await;
                         }
@@ -510,6 +528,7 @@ impl Session {
                         let core_dumped = map_err!(u8::decode(&mut r))? != 0;
                         let error_message = map_err!(String::decode(&mut r))?;
                         let lang_tag = map_err!(String::decode(&mut r))?;
+                        map_err!(ensure_end(&r))?;
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan
                                 .send(ChannelMsg::ExitSignal {
@@ -533,6 +552,7 @@ impl Session {
                     }
                     "keepalive@openssh.com" => {
                         let wants_reply = map_err!(u8::decode(&mut r))?;
+                        map_err!(ensure_end(&r))?;
                         if wants_reply == 1 {
                             if let Some(ref mut enc) = self.common.encrypted {
                                 trace!("Received channel keep alive message: {req:?}",);
@@ -570,6 +590,7 @@ impl Session {
             Some((&msg::CHANNEL_WINDOW_ADJUST, mut r)) => {
                 let channel_num = map_err!(ChannelId::decode(&mut r))?;
                 let amount = map_err!(u32::decode(&mut r))?;
+                map_err!(ensure_end(&r))?;
                 let mut new_size = 0;
                 debug!("channel_window_adjust amount: {amount:?}");
                 if let Some(ref mut enc) = self.common.encrypted {
@@ -601,6 +622,7 @@ impl Session {
                 let wants_reply = map_err!(u8::decode(&mut r))?;
                 if let Some(ref mut enc) = self.common.encrypted {
                     if req.starts_with("keepalive") {
+                        map_err!(ensure_end(&r))?;
                         if wants_reply == 1 {
                             trace!("Received keep alive message: {req:?}",);
                             self.common.wants_reply = false;
@@ -610,25 +632,18 @@ impl Session {
                         }
                     } else if req == "hostkeys-00@openssh.com" {
                         let mut keys = vec![];
-                        loop {
-                            match Bytes::decode(&mut r) {
-                                Ok(key) => {
-                                    let key = map_err!(parse_public_key(&key));
-                                    match key {
-                                        Ok(key) => keys.push(key),
-                                        Err(ref err) => {
-                                            debug!(
-                                                "failed to parse announced host key {key:?}: {err:?}",
-                                            )
-                                        }
-                                    }
-                                }
-                                Err(ssh_encoding::Error::Length) => break,
-                                x => {
-                                    map_err!(x)?;
+                        while !r.is_empty() {
+                            let key_blob = map_err!(Bytes::decode(&mut r))?;
+                            match parse_public_key(&key_blob) {
+                                Ok(key) => keys.push(key),
+                                Err(ref err) => {
+                                    debug!(
+                                        "failed to parse announced host key {key_blob:?}: {err:?}",
+                                    )
                                 }
                             }
                         }
+                        map_err!(ensure_end(&r))?;
                         return client.openssh_ext_host_keys_announced(keys, self).await;
                     } else {
                         warn!("Unhandled global request: {req:?} {wants_reply:?}",);
@@ -641,13 +656,15 @@ impl Session {
             }
             Some((&msg::CHANNEL_SUCCESS, mut r)) => {
                 let channel_num = map_err!(ChannelId::decode(&mut r))?;
+                map_err!(ensure_end(&r))?;
                 if let Some(chan) = self.channels.get(&channel_num) {
                     let _ = chan.send(ChannelMsg::Success).await;
                 }
                 client.channel_success(channel_num, self).await
             }
             Some((&msg::CHANNEL_FAILURE, mut r)) => {
                 let channel_num = map_err!(ChannelId::decode(&mut r))?;
+                map_err!(ensure_end(&r))?;
                 if let Some(chan) = self.channels.get(&channel_num) {
                     let _ = chan.send(ChannelMsg::Failure).await;
                 }
@@ -782,12 +799,15 @@ impl Session {
                 trace!("Global Request Success");
                 match self.open_global_requests.pop_front() {
                     Some(GlobalRequestResponse::Keepalive) => {
+                        map_err!(ensure_end(&r))?;
                         // ignore keepalives
                     }
                     Some(GlobalRequestResponse::Ping(return_channel)) => {
+                        map_err!(ensure_end(&r))?;
                         let _ = return_channel.send(());
                     }
                     Some(GlobalRequestResponse::NoMoreSessions) => {
+                        map_err!(ensure_end(&r))?;
                         debug!("no-more-sessions@openssh.com requests success");
                     }
                     Some(GlobalRequestResponse::TcpIpForward(return_channel)) => {
@@ -796,7 +816,16 @@ impl Session {
                             Some(0)
                         } else {
                             match u32::decode(&mut r) {
-                                Ok(port) => Some(port),
+                                Ok(port) => {
+                                    if let Err(e) = ensure_end(&r) {
+                                        error!(
+                                            "Error parsing port for TcpIpForward request: {e:?}"
+                                        );
+                                        None
+                                    } else {
+                                        Some(port)
+                                    }
+                                }
                                 Err(e) => {
                                     error!("Error parsing port for TcpIpForward request: {e:?}");
                                     None
@@ -806,12 +835,15 @@ impl Session {
                         let _ = return_channel.send(result);
                     }
                     Some(GlobalRequestResponse::CancelTcpIpForward(return_channel)) => {
+                        map_err!(ensure_end(&r))?;
                         let _ = return_channel.send(true);
                     }
                     Some(GlobalRequestResponse::StreamLocalForward(return_channel)) => {
+                        map_err!(ensure_end(&r))?;
                         let _ = return_channel.send(true);
                     }
                     Some(GlobalRequestResponse::CancelStreamLocalForward(return_channel)) => {
+                        map_err!(ensure_end(&r))?;
                         let _ = return_channel.send(true);
                     }
                     None => {
@@ -820,8 +852,9 @@ impl Session {
                 }
                 Ok(())
             }
-            Some((&msg::REQUEST_FAILURE, _)) => {
+            Some((&msg::REQUEST_FAILURE, r)) => {
                 trace!("global request failure");
+                map_err!(ensure_end(&r))?;
                 match self.open_global_requests.pop_front() {
                     Some(GlobalRequestResponse::Keepalive) => {
                         // ignore keepalives

@@ -32,7 +32,7 @@ use super::*;
 use crate::helpers::NameList;
 use crate::map_err;
 use crate::msg::SSH_OPEN_ADMINISTRATIVELY_PROHIBITED;
-use crate::parsing::{ChannelOpenConfirmation, ChannelType, OpenChannelMessage};
+use crate::parsing::{ChannelOpenConfirmation, ChannelType, OpenChannelMessage, ensure_end};
 
 impl Session {
     /// Returns false iff a request was rejected.
@@ -73,6 +73,7 @@ impl Session {
                 Some((&msg::SERVICE_REQUEST, mut r)),
             ) => {
                 let request = map_err!(String::decode(&mut r))?;
+                map_err!(ensure_end(&r))?;
                 debug!("request: {request:?}");
                 if request == "ssh-userauth" {
                     let auth_request = server_accept_service(
@@ -143,6 +144,165 @@ impl Session {
     }
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::tests::raw_no_crypto::{
+        MSG_SERVICE_REQUEST, MSG_USERAUTH_FAILURE, MSG_USERAUTH_REQUEST, RawSession,
+        assert_rejected, capture_panics, channel_request_payload, encode_string, pty_req_payload,
+        raw_auth_request_signal, raw_channel_request_signal, raw_service_request_signal,
+        read_packet, timeout,
+    };
+
+    #[tokio::test]
+    async fn malformed_pty_req_truncated_modes_rejected_by_server() {
+        let (result, panicked) = capture_panics(async {
+            timeout(raw_channel_request_signal(|server_channel| {
+                pty_req_payload(server_channel, &[Pty::VINTR as u8, 0, 0, 0])
+            }))
+            .await
+        })
+        .await;
+
+        assert!(!panicked, "truncated pty terminal modes caused a panic");
+        assert_rejected(result, "truncated pty terminal modes crashed or survived");
+    }
+
+    #[tokio::test]
+    async fn malformed_pty_req_rejects_bytes_after_mode_end() {
+        let result = timeout(raw_channel_request_signal(|server_channel| {
+            pty_req_payload(server_channel, &[Pty::TTY_OP_END as u8, 0])
+        }))
+        .await;
+
+        assert_rejected(
+            result,
+            "server accepted trailing bytes inside pty terminal modes",
+        );
+    }
+
+    #[tokio::test]
+    async fn malformed_pty_req_trailing_bytes_rejected_by_server() {
+        let result = timeout(raw_channel_request_signal(|server_channel| {
+            let mut payload = pty_req_payload(server_channel, &[Pty::TTY_OP_END as u8]);
+            payload.push(0);
+            payload
+        }))
+        .await;
+
+        assert_rejected(result, "server accepted a pty request with trailing bytes");
+    }
+
+    #[tokio::test]
+    async fn env_request_with_trailing_bytes_rejected_by_server() {
+        let result = timeout(raw_channel_request_signal(|server_channel| {
+            let mut payload = channel_request_payload(server_channel, b"env");
+            encode_string(&mut payload, b"LANG");
+            encode_string(&mut payload, b"C");
+            payload.push(0);
+            payload
+        }))
+        .await;
+
+        assert_rejected(result, "server accepted an env request with trailing bytes");
+    }
+
+    #[tokio::test]
+    async fn exec_request_with_trailing_bytes_rejected_by_server() {
+        let result = timeout(raw_channel_request_signal(|server_channel| {
+            let mut payload = channel_request_payload(server_channel, b"exec");
+            encode_string(&mut payload, b"true");
+            payload.push(0);
+            payload
+        }))
+        .await;
+
+        assert_rejected(result, "server accepted an exec request with trailing bytes");
+    }
+
+    #[tokio::test]
+    async fn signal_request_with_trailing_bytes_rejected_by_server() {
+        let result = timeout(raw_channel_request_signal(|server_channel| {
+            let mut payload = channel_request_payload(server_channel, b"signal");
+            encode_string(&mut payload, b"TERM");
+            payload.push(0);
+            payload
+        }))
+        .await;
+
+        assert_rejected(result, "server accepted a signal request with trailing bytes");
+    }
+
+    #[tokio::test]
+    async fn service_request_with_trailing_bytes_rejected_by_server() {
+        let result = timeout(raw_service_request_signal(|payload| {
+            payload.push(MSG_SERVICE_REQUEST);
+            encode_string(payload, b"ssh-userauth");
+            payload.push(0);
+        }))
+        .await;
+
+        assert_rejected(result, "server accepted a service request with trailing bytes");
+    }
+
+    #[tokio::test]
+    async fn auth_none_with_trailing_bytes_rejected_by_server() {
+        let result = timeout(raw_auth_request_signal(|payload| {
+            payload.push(MSG_USERAUTH_REQUEST);
+            encode_string(payload, b"test");
+            encode_string(payload, b"ssh-connection");
+            encode_string(payload, b"none");
+            payload.push(0);
+        }))
+        .await;
+
+        assert_rejected(result, "server accepted a none auth request with trailing bytes");
+    }
+
+    #[tokio::test]
+    async fn auth_password_with_trailing_bytes_rejected_by_server() {
+        let result = timeout(raw_auth_request_signal(|payload| {
+            payload.push(MSG_USERAUTH_REQUEST);
+            encode_string(payload, b"test");
+            encode_string(payload, b"ssh-connection");
+            encode_string(payload, b"password");
+            payload.push(0);
+            encode_string(payload, b"secret");
+            payload.push(0);
+        }))
+        .await;
+
+        assert_rejected(
+            result,
+            "server accepted a password auth request with trailing bytes",
+        );
+    }
+
+    #[tokio::test]
+    async fn password_change_request_is_parsed_and_rejected_by_server() {
+        let mut stream = RawSession::connect().await;
+        stream.service_request().await.unwrap();
+
+        let mut payload = Vec::new();
+        payload.push(MSG_USERAUTH_REQUEST);
+        encode_string(&mut payload, b"test");
+        encode_string(&mut payload, b"ssh-connection");
+        encode_string(&mut payload, b"password");
+        payload.push(1);
+        encode_string(&mut payload, b"old-password");
+        encode_string(&mut payload, b"new-password");
+        stream.send_packet(&payload).await.unwrap();
+
+        let failure = read_packet(&mut stream.stream).await.unwrap();
+        assert_eq!(failure.first(), Some(&MSG_USERAUTH_FAILURE));
+        assert!(
+            !stream.events.lock().unwrap().contains(&"auth_password"),
+            "password-change requests should not call normal password auth"
+        );
+        stream.server_task.abort();
+    }
+}
+
 fn server_accept_service(
     banner: Option<String>,
     methods: MethodSet,
@@ -196,9 +356,20 @@ impl Encrypted {
                 };
                 auth_user.clear();
                 auth_user.push_str(&user);
-                map_err!(u8::decode(r))?;
+                let change = map_err!(u8::decode(r))? != 0;
                 let password = map_err!(String::decode(r))?;
-                let auth = handler.auth_password(&user, &password).await?;
+                if change {
+                    let _new_password = map_err!(String::decode(r))?;
+                }
+                map_err!(ensure_end(r))?;
+                let auth = if change {
+                    Auth::Reject {
+                        proceed_with_methods: None,
+                        partial_success: false,
+                    }
+                } else {
+                    handler.auth_password(&user, &password).await?
+                };
                 if let Auth::Accept = auth {
                     server_auth_request_success(&mut self.write);
                     self.state = EncryptedState::InitCompression;
@@ -237,6 +408,7 @@ impl Encrypted {
                 };
 
                 until = initial_auth_until;
+                map_err!(ensure_end(r))?;
 
                 let auth = handler.auth_none(&user).await?;
                 if let Auth::Accept = auth {
@@ -269,6 +441,7 @@ impl Encrypted {
                 auth_user.push_str(&user);
                 let _ = map_err!(String::decode(r))?; // language_tag, deprecated.
                 let submethods = map_err!(String::decode(r))?;
+                map_err!(ensure_end(r))?;
                 debug!("{submethods:?}");
                 auth_request.current = Some(CurrentRequest::KeyboardInteractive {
                     submethods: submethods.to_string(),
@@ -374,7 +547,10 @@ impl Encrypted {
 
                     let encoded_signature = map_err!(Vec::<u8>::decode(r))?;
 
-                    let sig = map_err!(Signature::decode(&mut encoded_signature.as_slice()))?;
+                    let mut signature_reader = encoded_signature.as_slice();
+                    let sig = map_err!(Signature::decode(&mut signature_reader))?;
+                    map_err!(ensure_end(&signature_reader))?;
+                    map_err!(ensure_end(r))?;
 
                     let is_valid = if sent_pk_ok && user == auth_user {
                         true
@@ -433,6 +609,7 @@ impl Encrypted {
                     }
                     Ok(())
                 } else {
+                    map_err!(ensure_end(r))?;
                     auth_user.clear();
                     auth_user.push_str(user);
                     let auth = handler.auth_publickey_offered(user, &pubkey).await?;
@@ -532,6 +709,7 @@ async fn read_userauth_info_response<H: Handler + Send, R: Reader>(
         for _ in 0..n {
             responses.push(Bytes::decode(r).ok())
         }
+        map_err!(ensure_end(r))?;
 
         let auth = handler
             .auth_keyboard_interactive(user, submethods, Some(Response(&mut responses.into_iter())))
@@ -605,6 +783,7 @@ impl Session {
                 .map(|_| ()),
             msg::CHANNEL_CLOSE => {
                 let channel_num = map_err!(ChannelId::decode(r))?;
+                map_err!(ensure_end(r))?;
                 if let Some(ref mut enc) = self.common.encrypted {
                     enc.channels.remove(&channel_num);
                 }
@@ -620,6 +799,7 @@ impl Session {
             }
             msg::CHANNEL_EOF => {
                 let channel_num = map_err!(ChannelId::decode(r))?;
+                map_err!(ensure_end(r))?;
                 if let Some(chan) = self.channels.get(&channel_num) {
                     chan.send(ChannelMsg::Eof).await.unwrap_or(())
                 }
@@ -636,6 +816,7 @@ impl Session {
                 };
                 trace!("handler.data {ext:?} {channel_num:?}");
                 let data = map_err!(Bytes::decode(r))?;
+                map_err!(ensure_end(r))?;
                 let target = self.target_window_size;
 
                 if let Some(ref mut enc) = self.common.encrypted {
@@ -672,6 +853,7 @@ impl Session {
             msg::CHANNEL_WINDOW_ADJUST => {
                 let channel_num = map_err!(ChannelId::decode(r))?;
                 let amount = map_err!(u32::decode(r))?;
+                map_err!(ensure_end(r))?;
                 let mut new_size = 0;
                 if let Some(ref mut enc) = self.common.encrypted {
                     if let Some(channel) = enc.channels.get_mut(&channel_num) {
@@ -701,6 +883,7 @@ impl Session {
             msg::CHANNEL_OPEN_CONFIRMATION => {
                 debug!("channel_open_confirmation");
                 let msg = map_err!(ChannelOpenConfirmation::decode(r))?;
+                map_err!(ensure_end(r))?;
                 let local_id = ChannelId(msg.recipient_channel);
 
                 if let Some(ref mut enc) = self.common.encrypted {
@@ -756,14 +939,21 @@ impl Session {
                         let mut i = 0;
                         {
                             let mode_string = map_err!(Bytes::decode(r))?;
-                            while 5 * i < mode_string.len() {
+                            let mut mode_bytes = mode_string.as_ref();
+                            while !mode_bytes.is_empty() {
                                 #[allow(clippy::indexing_slicing)] // length checked
-                                let code = mode_string[5 * i];
+                                let code = mode_bytes[0];
                                 if code == 0 {
+                                    if mode_bytes.len() != 1 {
+                                        return Err(Error::Inconsistent.into());
+                                    }
                                     break;
                                 }
+                                if mode_bytes.len() < 5 {
+                                    return Err(Error::Inconsistent.into());
+                                }
                                 #[allow(clippy::indexing_slicing)] // length checked
-                                let num = BigEndian::read_u32(&mode_string[5 * i + 1..]);
+                                let num = BigEndian::read_u32(&mode_bytes[1..5]);
                                 debug!("code = {code:?}");
                                 if let Some(code) = Pty::from_u8(code) {
                                     #[allow(clippy::indexing_slicing)] // length checked
@@ -775,9 +965,11 @@ impl Session {
                                 } else {
                                     info!("pty-req: unknown pty code {code:?}");
                                 }
-                                i += 1
+                                i += 1;
+                                mode_bytes = &mode_bytes[5..];
                             }
                         }
+                        map_err!(ensure_end(r))?;
 
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan
@@ -813,6 +1005,7 @@ impl Session {
                         let x11_auth_protocol = map_err!(String::decode(r))?;
                         let x11_auth_cookie = map_err!(String::decode(r))?;
                         let x11_screen_number = map_err!(u32::decode(r))?;
+                        map_err!(ensure_end(r))?;
 
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan
@@ -840,6 +1033,7 @@ impl Session {
                     "env" => {
                         let env_variable = map_err!(String::decode(r))?;
                         let env_value = map_err!(String::decode(r))?;
+                        map_err!(ensure_end(r))?;
 
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan
@@ -857,6 +1051,7 @@ impl Session {
                             .await
                     }
                     "shell" => {
+                        map_err!(ensure_end(r))?;
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan
                                 .send(ChannelMsg::RequestShell { want_reply: true })
@@ -866,6 +1061,7 @@ impl Session {
                         handler.shell_request(channel_num, self).await
                     }
                     "auth-agent-req@openssh.com" => {
+                        map_err!(ensure_end(r))?;
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan
                                 .send(ChannelMsg::AgentForward { want_reply: true })
@@ -883,6 +1079,7 @@ impl Session {
                     }
                     "exec" => {
                         let req = map_err!(Bytes::decode(r))?;
+                        map_err!(ensure_end(r))?;
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan
                                 .send(ChannelMsg::Exec {
@@ -896,6 +1093,7 @@ impl Session {
                     }
                     "subsystem" => {
                         let name = map_err!(String::decode(r))?;
+                        map_err!(ensure_end(r))?;
 
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan
@@ -913,6 +1111,7 @@ impl Session {
                         let row_height = map_err!(u32::decode(r))?;
                         let pix_width = map_err!(u32::decode(r))?;
                         let pix_height = map_err!(u32::decode(r))?;
+                        map_err!(ensure_end(r))?;
 
                         if let Some(chan) = self.channels.get(&channel_num) {
                             let _ = chan
@@ -939,6 +1138,7 @@ impl Session {
                     }
                     "signal" => {
                         let signal = Sig::from_name(&map_err!(String::decode(r))?);
+                        map_err!(ensure_end(r))?;
                         if let Some(chan) = self.channels.get(&channel_num) {
                             chan.send(ChannelMsg::Signal {
                                 signal: signal.clone(),
@@ -963,6 +1163,7 @@ impl Session {
                     "tcpip-forward" => {
                         let address = map_err!(String::decode(r))?;
                         let port = map_err!(u32::decode(r))?;
+                        map_err!(ensure_end(r))?;
                         debug!("handler.tcpip_forward {address:?} {port:?}");
                         let mut returned_port = port;
                         let result = handler
@@ -985,6 +1186,7 @@ impl Session {
                     "cancel-tcpip-forward" => {
                         let address = map_err!(String::decode(r))?;
                         let port = map_err!(u32::decode(r))?;
+                        map_err!(ensure_end(r))?;
                         debug!("handler.cancel_tcpip_forward {address:?} {port:?}");
                         let result = handler.cancel_tcpip_forward(&address, port, self).await?;
                         if let Some(ref mut enc) = self.common.encrypted {
@@ -998,6 +1200,7 @@ impl Session {
                     }
                     "streamlocal-forward@openssh.com" => {
                         let server_socket_path = map_err!(String::decode(r))?;
+                        map_err!(ensure_end(r))?;
                         debug!("handler.streamlocal_forward {server_socket_path:?}");
                         let result = handler
                             .streamlocal_forward(&server_socket_path, self)
@@ -1013,6 +1216,7 @@ impl Session {
                     }
                     "cancel-streamlocal-forward@openssh.com" => {
                         let socket_path = map_err!(String::decode(r))?;
+                        map_err!(ensure_end(r))?;
                         debug!("handler.cancel_streamlocal_forward {socket_path:?}");
                         let result = handler
                             .cancel_streamlocal_forward(&socket_path, self)
@@ -1043,6 +1247,7 @@ impl Session {
                     .unwrap_or(ChannelOpenFailure::Unknown);
                 let description = map_err!(String::decode(r))?;
                 let language_tag = map_err!(String::decode(r))?;
+                map_err!(ensure_end(r))?;
 
                 trace!("Channel open failure description: {description}");
                 trace!("Channel open failure language tag: {language_tag}");
@@ -1064,9 +1269,11 @@ impl Session {
                 trace!("Global Request Success");
                 match self.open_global_requests.pop_front() {
                     Some(GlobalRequestResponse::Keepalive) => {
+                        map_err!(ensure_end(r))?;
                         // ignore keepalives
                     }
                     Some(GlobalRequestResponse::Ping(return_channel)) => {
+                        map_err!(ensure_end(r))?;
                         let _ = return_channel.send(());
                     }
                     Some(GlobalRequestResponse::TcpIpForward(return_channel)) => {
@@ -1075,7 +1282,16 @@ impl Session {
                             Some(0)
                         } else {
                             match u32::decode(r) {
-                                Ok(port) => Some(port),
+                                Ok(port) => {
+                                    if let Err(e) = ensure_end(r) {
+                                        error!(
+                                            "Error parsing port for TcpIpForward request: {e:?}"
+                                        );
+                                        None
+                                    } else {
+                                        Some(port)
+                                    }
+                                }
                                 Err(e) => {
                                     error!("Error parsing port for TcpIpForward request: {e:?}");
                                     None
@@ -1085,6 +1301,7 @@ impl Session {
                         let _ = return_channel.send(result);
                     }
                     Some(GlobalRequestResponse::CancelTcpIpForward(return_channel)) => {
+                        map_err!(ensure_end(r))?;
                         let _ = return_channel.send(true);
                     }
                     _ => {
@@ -1095,6 +1312,7 @@ impl Session {
             }
             msg::REQUEST_FAILURE => {
                 trace!("global request failure");
+                map_err!(ensure_end(r))?;
                 match self.open_global_requests.pop_front() {
                     Some(GlobalRequestResponse::Keepalive) => {
                         // ignore keepalives

@@ -14,6 +14,7 @@ use crate::kex::dh::groups::DhGroup;
 use crate::kex::{KEXES, KexAlgorithm, KexAlgorithmImplementor, KexCause, KexProgress};
 use crate::keys::key::parse_public_key;
 use crate::negotiation::{Names, Select};
+use crate::parsing::ensure_end;
 use crate::session::Exchange;
 use crate::sshbuffer::PacketWriter;
 use crate::{CryptoVec, Error, SshId, msg, negotiation, strict_kex_violation};
@@ -195,6 +196,7 @@ impl ClientKex {
 
                 let prime = Mpint::decode(&mut r)?;
                 let generator = Mpint::decode(&mut r)?;
+                ensure_end(&r)?;
                 debug!("received gex group: prime={prime}, generator={generator}");
 
                 let group = DhGroup {
@@ -287,7 +289,10 @@ impl ClientKex {
                 })?;
 
                 let signature = Bytes::decode(r)?;
-                let signature = Signature::decode(&mut &signature[..])?;
+                let mut signature_reader = &signature[..];
+                let signature = Signature::decode(&mut signature_reader)?;
+                ensure_end(&signature_reader)?;
+                ensure_end(r)?;
 
                 if let Err(e) =
                     signature::Verifier::verify(&server_host_key, hash.as_ref(), &signature)
@@ -338,6 +343,8 @@ impl ClientKex {
                     );
                     return Err(Error::Kex);
                 }
+                let r = &input.buffer[1..];
+                ensure_end(&r)?;
 
                 Ok(KexProgress::Done {
                     newkeys,

@@ -26,6 +26,7 @@ use crate::kex::{
     EXTENSION_OPENSSH_STRICT_KEX_AS_CLIENT, EXTENSION_OPENSSH_STRICT_KEX_AS_SERVER, KexCause,
 };
 use crate::keys::key::safe_rng;
+use crate::parsing::ensure_end;
 #[cfg(not(target_arch = "wasm32"))]
 use crate::server::Config;
 use crate::sshbuffer::PacketWriter;
@@ -343,6 +344,8 @@ pub(crate) trait Select {
         String::decode(&mut r)?; // languages server-to-client
 
         let follows = u8::decode(&mut r)? != 0;
+        u32::decode(&mut r)?;
+        ensure_end(&r)?;
         Ok(Names {
             kex: kex_algorithm,
             key: key_algorithm,

@@ -4,6 +4,23 @@ use crate::msg;
 
 use crate::map_err;
 
+/// Require a decoded known-message payload to be fully consumed.
+///
+/// SSH RFCs and implemented OpenSSH extensions define exact field layouts for
+/// known message types. Callers use this after decoding those fields so
+/// malformed packets with trailing payload bytes are rejected instead of being
+/// treated as canonical messages.
+pub(crate) fn ensure_end(reader: &impl Reader) -> Result<(), crate::Error> {
+    if reader.is_finished() {
+        Ok(())
+    } else {
+        Err(ssh_encoding::Error::TrailingData {
+            remaining: reader.remaining_len(),
+        }
+        .into())
+    }
+}
+
 #[derive(Debug)]
 pub struct OpenChannelMessage {
     pub typ: ChannelType,
@@ -13,6 +30,12 @@ pub struct OpenChannelMessage {
 }
 
 impl OpenChannelMessage {
+    /// Parse an SSH `CHANNEL_OPEN` payload.
+    ///
+    /// Known channel types are parsed according to their fixed layouts and must
+    /// not contain trailing bytes. Unknown extension channel types remain
+    /// intentionally opaque so applications can implement extension-specific
+    /// parsing and compatibility behavior.
     pub fn parse<R: Reader>(r: &mut R) -> Result<Self, crate::Error> {
         // https://tools.ietf.org/html/rfc4254#section-5.1
         let typ = map_err!(String::decode(r))?;
@@ -21,24 +44,46 @@ impl OpenChannelMessage {
         let maxpacket = map_err!(u32::decode(r))?;
 
         let typ = match typ.as_str() {
-            "session" => ChannelType::Session,
+            "session" => {
+                ensure_end(r)?;
+                ChannelType::Session
+            }
             "x11" => {
                 let originator_address = map_err!(String::decode(r))?;
                 let originator_port = map_err!(u32::decode(r))?;
+                ensure_end(r)?;
                 ChannelType::X11 {
                     originator_address,
                     originator_port,
                 }
             }
-            "direct-tcpip" => ChannelType::DirectTcpip(TcpChannelInfo::decode(r)?),
+            "direct-tcpip" => {
+                let info = TcpChannelInfo::decode(r)?;
+                ensure_end(r)?;
+                ChannelType::DirectTcpip(info)
+            }
             "direct-streamlocal@openssh.com" => {
-                ChannelType::DirectStreamLocal(StreamLocalChannelInfo::decode(r)?)
+                let info = StreamLocalChannelInfo::decode(r)?;
+                String::decode(r)?; // originator address/reserved
+                u32::decode(r)?; // originator port/reserved
+                ensure_end(r)?;
+                ChannelType::DirectStreamLocal(info)
+            }
+            "forwarded-tcpip" => {
+                let info = TcpChannelInfo::decode(r)?;
+                ensure_end(r)?;
+                ChannelType::ForwardedTcpIp(info)
             }
-            "forwarded-tcpip" => ChannelType::ForwardedTcpIp(TcpChannelInfo::decode(r)?),
             "forwarded-streamlocal@openssh.com" => {
-                ChannelType::ForwardedStreamLocal(StreamLocalChannelInfo::decode(r)?)
+                let info = StreamLocalChannelInfo::decode(r)?;
+                String::decode(r)?; // reserved
+                ensure_end(r)?;
+                ChannelType::ForwardedStreamLocal(info)
+            }
+            "auth-agent@openssh.com" => {
+                ensure_end(r)?;
+                ChannelType::AgentForward
             }
-            "auth-agent@openssh.com" => ChannelType::AgentForward,
             _ => ChannelType::Unknown { typ },
         };
 
@@ -177,3 +222,51 @@ impl Decode for ChannelOpenConfirmation {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{ChannelType, OpenChannelMessage};
+    use crate::tests::raw_no_crypto::{channel_open_payload, encode_string, push_u32};
+
+    #[test]
+    fn known_channel_open_with_trailing_bytes_is_rejected() {
+        let mut payload = channel_open_payload(b"session");
+        payload.push(0);
+
+        assert!(
+            OpenChannelMessage::parse(&mut payload.as_slice()).is_err(),
+            "known channel-open type accepted trailing bytes"
+        );
+    }
+
+    #[test]
+    fn unknown_channel_open_with_extra_payload_stays_permissive() {
+        let mut payload = channel_open_payload(b"unknown@example.com");
+        payload.extend_from_slice(b"opaque");
+
+        let parsed = OpenChannelMessage::parse(&mut payload.as_slice())
+            .expect("unknown channel-open payload should remain opaque");
+
+        assert!(matches!(parsed.typ, ChannelType::Unknown { .. }));
+    }
+
+    #[test]
+    fn openssh_streamlocal_channel_open_reserved_fields_are_consumed() {
+        let mut direct = channel_open_payload(b"direct-streamlocal@openssh.com");
+        encode_string(&mut direct, b"/tmp/socket");
+        encode_string(&mut direct, b"");
+        push_u32(&mut direct, 0);
+
+        let parsed = OpenChannelMessage::parse(&mut direct.as_slice())
+            .expect("direct streamlocal reserved fields should be consumed");
+        assert!(matches!(parsed.typ, ChannelType::DirectStreamLocal(_)));
+
+        let mut forwarded = channel_open_payload(b"forwarded-streamlocal@openssh.com");
+        encode_string(&mut forwarded, b"/tmp/socket");
+        encode_string(&mut forwarded, b"");
+
+        let parsed = OpenChannelMessage::parse(&mut forwarded.as_slice())
+            .expect("forwarded streamlocal reserved field should be consumed");
+        assert!(matches!(parsed.typ, ChannelType::ForwardedStreamLocal(_)));
+    }
+}

@@ -13,6 +13,7 @@ use crate::kex::dh::biguint_to_mpint;
 use crate::kex::{KexAlgorithm, KexAlgorithmImplementor, KexCause, KEXES};
 use crate::keys::key::PrivateKeyWithHashAlg;
 use crate::negotiation::{is_key_compatible_with_algo, Names, Select};
+use crate::parsing::ensure_end;
 use crate::{msg, negotiation};
 
 thread_local! {
@@ -173,7 +174,9 @@ impl ServerKex {
                 }
 
                 #[allow(clippy::indexing_slicing)] // length checked
-                let gex_params = GexParams::decode(&mut &input.buffer[1..])?;
+                let mut r = &input.buffer[1..];
+                let gex_params = GexParams::decode(&mut r)?;
+                ensure_end(&r)?;
                 debug!("client requests a gex group: {gex_params:?}");
 
                 let Some(dh_group) = handler.lookup_dh_gex_group(&gex_params).await? else {
@@ -236,6 +239,7 @@ impl ServerKex {
                 self.exchange
                     .client_ephemeral
                     .extend_from_slice(&Bytes::decode(&mut r).map_err(Into::into)?);
+                ensure_end(&r)?;
 
                 let exchange = &mut self.exchange;
                 kex.server_dh(exchange, &input.buffer)?;
@@ -324,6 +328,8 @@ impl ServerKex {
                     );
                     return Err(Error::Kex.into());
                 }
+                let r = &input.buffer[1..];
+                ensure_end(&r)?;
 
                 debug!("new keys received");
                 Ok(KexProgress::Done {
@@ -372,3 +378,21 @@ fn compute_keys(
         session_id: session_id_cv,
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::tests::raw_no_crypto::{
+        assert_rejected, kexinit_payload, raw_kex_signal, timeout,
+    };
+
+    #[tokio::test]
+    async fn kexinit_with_trailing_bytes_rejected_by_server() {
+        let result = timeout(raw_kex_signal(|payload| {
+            payload.extend_from_slice(&kexinit_payload("none"));
+            payload.push(0);
+        }))
+        .await;
+
+        assert_rejected(result, "server accepted a kexinit with trailing bytes");
+    }
+}

@@ -714,6 +714,458 @@ mod server_kex_junk {
     }
 }
 
+pub(crate) mod raw_no_crypto {
+    use std::borrow::Cow;
+    use std::io;
+    use std::sync::{Arc, Mutex, OnceLock};
+    use std::time::Duration;
+
+    use byteorder::{BigEndian, ByteOrder};
+    use ssh_key::{Algorithm, PrivateKey};
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    use super::*;
+
+    pub(crate) const MSG_SERVICE_REQUEST: u8 = 5;
+    pub(crate) const MSG_SERVICE_ACCEPT: u8 = 6;
+    pub(crate) const MSG_KEXINIT: u8 = 20;
+    pub(crate) const MSG_NEWKEYS: u8 = 21;
+    pub(crate) const MSG_USERAUTH_REQUEST: u8 = 50;
+    pub(crate) const MSG_USERAUTH_FAILURE: u8 = 51;
+    pub(crate) const MSG_USERAUTH_SUCCESS: u8 = 52;
+    pub(crate) const MSG_CHANNEL_OPEN: u8 = 90;
+    pub(crate) const MSG_CHANNEL_OPEN_CONFIRMATION: u8 = 91;
+    pub(crate) const MSG_CHANNEL_REQUEST: u8 = 98;
+
+    pub(crate) async fn raw_service_request_signal(
+        build_payload: impl FnOnce(&mut Vec<u8>),
+    ) -> ServerSignal {
+        let mut stream = RawSession::connect().await;
+        let mut payload = Vec::new();
+        build_payload(&mut payload);
+        stream.send_packet(&payload).await.unwrap();
+        stream.result().await
+    }
+
+    pub(crate) async fn raw_auth_request_signal(
+        build_payload: impl FnOnce(&mut Vec<u8>),
+    ) -> ServerSignal {
+        let mut stream = RawSession::connect().await;
+        stream.service_request().await.unwrap();
+
+        let mut payload = Vec::new();
+        build_payload(&mut payload);
+        stream.send_packet(&payload).await.unwrap();
+        stream.result().await
+    }
+
+    pub(crate) async fn raw_kex_signal(build_payload: impl FnOnce(&mut Vec<u8>)) -> ServerSignal {
+        let mut stream = RawSession::connect_without_kex().await;
+
+        let mut payload = Vec::new();
+        build_payload(&mut payload);
+        stream.send_packet(&payload).await.unwrap();
+        stream.result().await
+    }
+
+    pub(crate) async fn raw_channel_request_signal(
+        build_payload: impl FnOnce(u32) -> Vec<u8>,
+    ) -> ServerSignal {
+        let mut stream = RawSession::connect().await;
+        stream.auth_none().await.unwrap();
+        let server_channel = stream.open_session().await.unwrap();
+        stream
+            .send_packet(&build_payload(server_channel))
+            .await
+            .unwrap();
+        stream.result().await
+    }
+
+    pub(crate) struct RawSession {
+        pub(crate) events: Arc<Mutex<Vec<&'static str>>>,
+        pub(crate) stream: tokio::net::TcpStream,
+        pub(crate) server_task: tokio::task::JoinHandle<Result<(), Error>>,
+    }
+
+    impl RawSession {
+        pub(crate) async fn connect() -> Self {
+            let mut stream = Self::connect_without_kex().await;
+            raw_client_no_crypto_kex(&mut stream.stream).await.unwrap();
+            stream
+        }
+
+        pub(crate) async fn connect_without_kex() -> Self {
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let addr = listener.local_addr().unwrap();
+            let events = Arc::new(Mutex::new(Vec::new()));
+            let server_events = events.clone();
+            let server_task = tokio::spawn(async move {
+                let (socket, _) = listener.accept().await.unwrap();
+                let running =
+                    server::run_stream(
+                        no_crypto_server_config(),
+                        socket,
+                        MalformedInputServer {
+                            events: server_events,
+                        },
+                    )
+                    .await
+                    .unwrap();
+                running.await
+            });
+
+            let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
+            stream.write_all(b"SSH-2.0-russh-test\r\n").await.unwrap();
+            read_ssh_id(&mut stream).await.unwrap();
+            let _server_kex = read_packet(&mut stream).await.unwrap();
+            Self {
+                events,
+                stream,
+                server_task,
+            }
+        }
+
+        pub(crate) async fn service_request(&mut self) -> io::Result<()> {
+            let mut service = Vec::new();
+            service.push(MSG_SERVICE_REQUEST);
+            encode_string(&mut service, b"ssh-userauth");
+            self.send_packet(&service).await?;
+
+            let accept = read_packet(&mut self.stream).await?;
+            assert_eq!(accept.first(), Some(&MSG_SERVICE_ACCEPT));
+            Ok(())
+        }
+
+        pub(crate) async fn auth_none(&mut self) -> io::Result<()> {
+            self.service_request().await?;
+
+            let mut auth = Vec::new();
+            auth.push(MSG_USERAUTH_REQUEST);
+            encode_string(&mut auth, b"test");
+            encode_string(&mut auth, b"ssh-connection");
+            encode_string(&mut auth, b"none");
+            self.send_packet(&auth).await?;
+
+            let success = read_packet(&mut self.stream).await?;
+            assert_eq!(success.first(), Some(&MSG_USERAUTH_SUCCESS));
+            Ok(())
+        }
+
+        pub(crate) async fn open_session(&mut self) -> io::Result<u32> {
+            let mut open = Vec::new();
+            open.push(MSG_CHANNEL_OPEN);
+            encode_string(&mut open, b"session");
+            push_u32(&mut open, 0);
+            push_u32(&mut open, 1024 * 1024);
+            push_u32(&mut open, 32 * 1024);
+            self.send_packet(&open).await?;
+
+            let confirmation = read_packet(&mut self.stream).await?;
+            assert_eq!(confirmation.first(), Some(&MSG_CHANNEL_OPEN_CONFIRMATION));
+            Ok(BigEndian::read_u32(&confirmation[5..9]))
+        }
+
+        pub(crate) async fn send_packet(&mut self, payload: &[u8]) -> io::Result<()> {
+            self.stream.write_all(&ssh_packet(payload)).await?;
+            self.stream.flush().await
+        }
+
+        pub(crate) async fn result(mut self) -> ServerSignal {
+            let result =
+                tokio::time::timeout(Duration::from_millis(200), &mut self.server_task).await;
+            let events = self.events.lock().unwrap().clone();
+
+            match result {
+                Ok(Ok(Ok(()))) => ServerSignal::Closed { events },
+                Ok(Ok(Err(_error))) => ServerSignal::ProtocolError { events },
+                Ok(Err(join)) if join.is_panic() => ServerSignal::Panicked { events },
+                Err(_) => {
+                    self.server_task.abort();
+                    ServerSignal::Survived { events }
+                }
+                _ => ServerSignal::Closed { events },
+            }
+        }
+    }
+
+    fn no_crypto_server_config() -> Arc<server::Config> {
+        let mut config = server::Config::default();
+        config.inactivity_timeout = None;
+        config.auth_rejection_time = Duration::from_millis(1);
+        config.auth_rejection_time_initial = Some(Duration::from_millis(1));
+        config.preferred = no_crypto_preferred();
+        config
+            .keys
+            .push(PrivateKey::random(&mut rand::rng(), Algorithm::Ed25519).unwrap());
+        Arc::new(config)
+    }
+
+    fn no_crypto_preferred() -> Preferred {
+        Preferred {
+            kex: Cow::Owned(vec![kex::NONE]),
+            key: Cow::Owned(vec![Algorithm::Ed25519]),
+            cipher: Cow::Owned(vec![cipher::NONE]),
+            mac: Cow::Owned(vec![mac::NONE]),
+            compression: Cow::Owned(vec![compression::NONE]),
+        }
+    }
+
+    async fn raw_client_no_crypto_kex(stream: &mut tokio::net::TcpStream) -> io::Result<()> {
+        stream
+            .write_all(&ssh_packet(&kexinit_payload("none")))
+            .await?;
+        let newkeys = read_packet(stream).await?;
+        assert_eq!(newkeys.first(), Some(&MSG_NEWKEYS));
+        stream.write_all(&ssh_packet(&[MSG_NEWKEYS])).await?;
+        stream.flush().await
+    }
+
+    pub(crate) fn pty_req_payload(server_channel: u32, terminal_modes: &[u8]) -> Vec<u8> {
+        let mut payload = channel_request_payload(server_channel, b"pty-req");
+        encode_string(&mut payload, b"xterm");
+        push_u32(&mut payload, 80);
+        push_u32(&mut payload, 24);
+        push_u32(&mut payload, 0);
+        push_u32(&mut payload, 0);
+        encode_string(&mut payload, terminal_modes);
+        payload
+    }
+
+    pub(crate) fn channel_open_payload(channel_type: &[u8]) -> Vec<u8> {
+        let mut payload = Vec::new();
+        encode_string(&mut payload, channel_type);
+        push_u32(&mut payload, 0);
+        push_u32(&mut payload, 1024 * 1024);
+        push_u32(&mut payload, 32 * 1024);
+        payload
+    }
+
+    pub(crate) fn channel_request_payload(server_channel: u32, request_type: &[u8]) -> Vec<u8> {
+        let mut payload = Vec::new();
+        payload.push(MSG_CHANNEL_REQUEST);
+        push_u32(&mut payload, server_channel);
+        encode_string(&mut payload, request_type);
+        payload.push(1);
+        payload
+    }
+
+    pub(crate) fn kexinit_payload(kex_name: &str) -> Vec<u8> {
+        let mut payload = Vec::new();
+        payload.push(MSG_KEXINIT);
+        payload.extend_from_slice(&[0; 16]);
+        encode_name_list(&mut payload, &[kex_name]);
+        encode_name_list(&mut payload, &["ssh-ed25519"]);
+        encode_name_list(&mut payload, &["none"]);
+        encode_name_list(&mut payload, &["none"]);
+        encode_name_list(&mut payload, &["none"]);
+        encode_name_list(&mut payload, &["none"]);
+        encode_name_list(&mut payload, &["none"]);
+        encode_name_list(&mut payload, &["none"]);
+        encode_name_list(&mut payload, &[]);
+        encode_name_list(&mut payload, &[]);
+        payload.push(0);
+        push_u32(&mut payload, 0);
+        payload
+    }
+
+    fn ssh_packet(payload: &[u8]) -> Vec<u8> {
+        let mut padding_len = 8 - ((5 + payload.len()) % 8);
+        if padding_len < 4 {
+            padding_len += 8;
+        }
+        let packet_len = 1 + payload.len() + padding_len;
+        let mut packet = Vec::with_capacity(4 + packet_len);
+        push_u32(&mut packet, packet_len as u32);
+        packet.push(padding_len as u8);
+        packet.extend_from_slice(payload);
+        packet.resize(packet.len() + padding_len, 0);
+        packet
+    }
+
+    pub(crate) async fn read_packet(stream: &mut tokio::net::TcpStream) -> io::Result<Vec<u8>> {
+        let mut len_buf = [0; 4];
+        stream.read_exact(&mut len_buf).await?;
+        let packet_len = BigEndian::read_u32(&len_buf) as usize;
+        let mut packet = vec![0; packet_len];
+        stream.read_exact(&mut packet).await?;
+        let padding_len = packet[0] as usize;
+        Ok(packet[1..packet.len() - padding_len].to_vec())
+    }
+
+    async fn read_ssh_id(stream: &mut tokio::net::TcpStream) -> io::Result<Vec<u8>> {
+        let mut id = Vec::new();
+        loop {
+            let mut byte = [0];
+            stream.read_exact(&mut byte).await?;
+            id.push(byte[0]);
+            if byte[0] == b'\n' {
+                return Ok(id);
+            }
+        }
+    }
+
+    fn encode_name_list(buf: &mut Vec<u8>, names: &[&str]) {
+        encode_string(buf, names.join(",").as_bytes());
+    }
+
+    pub(crate) fn encode_string(buf: &mut Vec<u8>, value: &[u8]) {
+        push_u32(buf, value.len() as u32);
+        buf.extend_from_slice(value);
+    }
+
+    pub(crate) fn push_u32(buf: &mut Vec<u8>, value: u32) {
+        let mut bytes = [0; 4];
+        BigEndian::write_u32(&mut bytes, value);
+        buf.extend_from_slice(&bytes);
+    }
+
+    pub(crate) async fn timeout(
+        future: impl Future<Output = ServerSignal>,
+    ) -> Result<ServerSignal, tokio::time::error::Elapsed> {
+        tokio::time::timeout(Duration::from_secs(3), future).await
+    }
+
+    pub(crate) async fn capture_panics<T>(future: impl Future<Output = T>) -> (T, bool) {
+        static PANIC_HOOK_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
+
+        let _guard = PANIC_HOOK_LOCK
+            .get_or_init(|| Mutex::new(()))
+            .lock()
+            .unwrap();
+        let panicked = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let panicked_hook = panicked.clone();
+        let previous_hook = std::panic::take_hook();
+        std::panic::set_hook(Box::new(move |_| {
+            panicked_hook.store(true, std::sync::atomic::Ordering::SeqCst);
+        }));
+
+        let result = future.await;
+
+        std::panic::set_hook(previous_hook);
+        (result, panicked.load(std::sync::atomic::Ordering::SeqCst))
+    }
+
+    #[derive(Debug)]
+    pub(crate) enum ServerSignal {
+        Closed { events: Vec<&'static str> },
+        ProtocolError { events: Vec<&'static str> },
+        Panicked { events: Vec<&'static str> },
+        Survived { events: Vec<&'static str> },
+    }
+
+    impl ServerSignal {
+        pub(crate) fn events(&self) -> &[&'static str] {
+            match self {
+                Self::Closed { events }
+                | Self::ProtocolError { events }
+                | Self::Panicked { events }
+                | Self::Survived { events } => events,
+            }
+        }
+    }
+
+    pub(crate) fn assert_rejected(
+        result: Result<ServerSignal, tokio::time::error::Elapsed>,
+        message: &str,
+    ) {
+        assert!(
+            matches!(
+                result,
+                Ok(ServerSignal::Closed { .. } | ServerSignal::ProtocolError { .. })
+            ),
+            "{message}: {result:?}; handler events: {:?}",
+            result.as_ref().ok().map(ServerSignal::events).unwrap_or(&[])
+        );
+    }
+
+    #[derive(Clone)]
+    struct MalformedInputServer {
+        events: Arc<Mutex<Vec<&'static str>>>,
+    }
+
+    impl MalformedInputServer {
+        fn record(&self, event: &'static str) {
+            self.events.lock().unwrap().push(event);
+        }
+    }
+
+    impl server::Handler for MalformedInputServer {
+        type Error = Error;
+
+        async fn auth_none(&mut self, _user: &str) -> Result<server::Auth, Self::Error> {
+            self.record("auth_none");
+            Ok(server::Auth::Accept)
+        }
+
+        async fn auth_password(
+            &mut self,
+            _user: &str,
+            _password: &str,
+        ) -> Result<server::Auth, Self::Error> {
+            self.record("auth_password");
+            Ok(server::Auth::Reject {
+                proceed_with_methods: None,
+                partial_success: false,
+            })
+        }
+
+        async fn channel_open_session(
+            &mut self,
+            _channel: Channel<server::Msg>,
+            _session: &mut server::Session,
+        ) -> Result<bool, Self::Error> {
+            self.record("channel_open_session");
+            Ok(true)
+        }
+
+        async fn pty_request(
+            &mut self,
+            _channel: ChannelId,
+            _term: &str,
+            _col_width: u32,
+            _row_height: u32,
+            _pix_width: u32,
+            _pix_height: u32,
+            _modes: &[(Pty, u32)],
+            _session: &mut server::Session,
+        ) -> Result<(), Self::Error> {
+            self.record("pty_request");
+            Ok(())
+        }
+
+        async fn env_request(
+            &mut self,
+            _channel: ChannelId,
+            _variable_name: &str,
+            _variable_value: &str,
+            _session: &mut server::Session,
+        ) -> Result<(), Self::Error> {
+            self.record("env_request");
+            Ok(())
+        }
+
+        async fn exec_request(
+            &mut self,
+            _channel: ChannelId,
+            _data: &[u8],
+            _session: &mut server::Session,
+        ) -> Result<(), Self::Error> {
+            self.record("exec_request");
+            Ok(())
+        }
+
+        async fn signal(
+            &mut self,
+            _channel: ChannelId,
+            _signal: Sig,
+            _session: &mut server::Session,
+        ) -> Result<(), Self::Error> {
+            self.record("signal");
+            Ok(())
+        }
+    }
+}
+
 /// Integration test for FutureCertificate authentication flow
 #[cfg(unix)]
 mod future_certificate {