Bump release candidate to rc7

Cargo.lock

@@ -436,7 +436,7 @@ dependencies = [
 
 [[package]]
 name = "dosh"
-version = "1.0.0-rc3"
+version = "1.0.0-rc7"
 dependencies = [
  "anyhow",
  "base64",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dosh"
-version = "1.0.0-rc3"
+version = "1.0.0-rc7"
 edition = "2024"
 license = "MIT"
 

src/bin/dosh-client.rs

@@ -66,6 +66,8 @@ const STARTUP_INPUT_HOLD: Duration = Duration::from_millis(750);
 const POST_SUBMIT_ALL_INPUT_HOLD: Duration = Duration::from_millis(120);
 const STALE_TERMINAL_INPUT_AFTER: Duration = Duration::from_secs(2);
 const POST_RECONNECT_STALE_INPUT_GRACE: Duration = Duration::from_secs(2);
+const FOCUS_REPAINT_COOLDOWN: Duration = Duration::from_secs(1);
+const ALT_SCREEN_IDLE_REPAINT_AFTER: Duration = Duration::from_secs(15);
 
 /// Sentinel `target_host` the server uses on a server-initiated `StreamOpen` that
 /// represents an SSH-agent connection (rather than a TCP target). The client
@@ -4334,10 +4336,14 @@ async fn run_terminal(
     let mut startup_input_hold_until: Option<Instant> = None;
     let mut startup_gate_mode = StartupGateMode::HoldControl;
     let mut stale_terminal_input_suppress_until: Option<Instant> = None;
+    let mut last_terminal_frame_at = Instant::now();
+    let mut last_focus_repaint_at = Instant::now() - FOCUS_REPAINT_COOLDOWN;
+    let mut last_idle_repaint_attempt_at = Instant::now() - ALT_SCREEN_IDLE_REPAINT_AFTER;
     if let Some(frame) = first_frame {
         if !forward_only {
             render_frame(&frame)?;
             predictor.observe_output(&frame.bytes);
+            last_terminal_frame_at = Instant::now();
         }
         if frame.closed {
             return Ok(());
@@ -4395,6 +4401,46 @@ async fn run_terminal(
                         if input_matches_escape(&bytes, escape_key.as_deref()) {
                             break;
                         }
+                        let saw_focus_in = input_contains_focus_in(&bytes);
+                        if !forward_only
+                            && saw_focus_in
+                            && last_focus_repaint_at.elapsed() >= FOCUS_REPAINT_COOLDOWN
+                        {
+                            last_focus_repaint_at = Instant::now();
+                            if let Some(frame) = reconnect(
+                                &socket,
+                                &mut cred,
+                                &mut send_seq,
+                                last_size,
+                                &mut frame_buffer,
+                                &mut predictor,
+                            )
+                            .await?
+                            {
+                                refresh_live_addr(&mut addr, &cred)?;
+                                arm_stale_terminal_input_suppression(
+                                    &mut stale_terminal_input_suppress_until,
+                                );
+                                render_frame(&frame)?;
+                                predictor.observe_output(&frame.bytes);
+                                last_terminal_frame_at = Instant::now();
+                                last_packet_at = Instant::now();
+                                flush_pending_user_input(
+                                    &socket,
+                                    addr,
+                                    &cred,
+                                    &mut send_seq,
+                                    &mut predictor,
+                                    &mut pending_user_input,
+                                    &mut pending_user_input_bytes,
+                                )
+                                .await?;
+                            }
+                        }
+                        bytes = strip_terminal_focus_reports(&bytes);
+                        if bytes.is_empty() {
+                            continue;
+                        }
                         if should_strip_stale_terminal_reports(
                             last_packet_at.elapsed(),
                             stale_terminal_input_suppress_until,
@@ -4508,6 +4554,7 @@ async fn run_terminal(
                                     if !forward_only {
                                         render_frame(&frame)?;
                                         predictor.observe_output(&frame.bytes);
+                                        last_terminal_frame_at = Instant::now();
                                     }
                                     last_packet_at = Instant::now();
                                     flush_pending_user_input(
@@ -4570,6 +4617,7 @@ async fn run_terminal(
                     if !forward_only {
                         render_frame(&frame)?;
                         predictor.observe_output(&frame.bytes);
+                        last_terminal_frame_at = Instant::now();
                     }
                     last_packet_at = Instant::now();
                     flush_pending_user_input(
@@ -4623,6 +4671,7 @@ async fn run_terminal(
                                 if !forward_only {
                                     render_frame(&frame)?;
                                     predictor.observe_output(&frame.bytes);
+                                    last_terminal_frame_at = Instant::now();
                                 }
                                 last_packet_at = Instant::now();
                                 flush_pending_user_input(
@@ -4648,6 +4697,7 @@ async fn run_terminal(
                             if !forward_only {
                                 render_frame(&frame)?;
                                 predictor.observe_output(&frame.bytes);
+                                last_terminal_frame_at = Instant::now();
                                 flush_startup_input_if_ready(
                                     &socket,
                                     addr,
@@ -4747,6 +4797,7 @@ async fn run_terminal(
                                 if !forward_only {
                                     render_frame(&frame)?;
                                     predictor.observe_output(&frame.bytes);
+                                    last_terminal_frame_at = Instant::now();
                                 }
                                 last_packet_at = Instant::now();
                                 flush_pending_user_input(
@@ -5142,6 +5193,7 @@ async fn run_terminal(
                         if !forward_only {
                             render_frame(&frame)?;
                             predictor.observe_output(&frame.bytes);
+                            last_terminal_frame_at = Instant::now();
                         }
                         last_packet_at = Instant::now();
                         flush_pending_user_input(
@@ -5183,6 +5235,44 @@ async fn run_terminal(
                         (&mut startup_input_hold_until, &mut startup_gate_mode),
                     )
                     .await?;
+                    let now = Instant::now();
+                    if should_repaint_idle_alternate_screen(
+                        predictor.alternate_screen,
+                        last_terminal_frame_at,
+                        last_idle_repaint_attempt_at,
+                        now,
+                    ) {
+                        last_idle_repaint_attempt_at = now;
+                        if let Some(frame) = reconnect(
+                            &socket,
+                            &mut cred,
+                            &mut send_seq,
+                            last_size,
+                            &mut frame_buffer,
+                            &mut predictor,
+                        )
+                        .await?
+                        {
+                            refresh_live_addr(&mut addr, &cred)?;
+                            arm_stale_terminal_input_suppression(
+                                &mut stale_terminal_input_suppress_until,
+                            );
+                            render_frame(&frame)?;
+                            predictor.observe_output(&frame.bytes);
+                            last_terminal_frame_at = Instant::now();
+                            last_packet_at = Instant::now();
+                            flush_pending_user_input(
+                                &socket,
+                                addr,
+                                &cred,
+                                &mut send_seq,
+                                &mut predictor,
+                                &mut pending_user_input,
+                                &mut pending_user_input_bytes,
+                            )
+                            .await?;
+                        }
+                    }
                 }
                 // Refresh / clear the non-destructive disconnect status line based
                 // on how long the link has been silent (recomputed after any
@@ -5552,6 +5642,41 @@ fn should_strip_stale_terminal_reports(
         || suppress_until.is_some_and(|deadline| Instant::now() < deadline)
 }
 
+fn input_contains_focus_in(bytes: &[u8]) -> bool {
+    contains_bytes(bytes, b"\x1b[I") || contains_bytes(bytes, b"\x9bI")
+}
+
+fn strip_terminal_focus_reports(bytes: &[u8]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(bytes.len());
+    let mut offset = 0;
+    while offset < bytes.len() {
+        match bytes.get(offset..) {
+            Some([0x1b, b'[', b'I' | b'O', ..]) => {
+                offset += 3;
+            }
+            Some([0x9b, b'I' | b'O', ..]) => {
+                offset += 2;
+            }
+            _ => {
+                out.push(bytes[offset]);
+                offset += 1;
+            }
+        }
+    }
+    out
+}
+
+fn should_repaint_idle_alternate_screen(
+    alternate_screen: bool,
+    last_terminal_frame_at: Instant,
+    last_attempt_at: Instant,
+    now: Instant,
+) -> bool {
+    alternate_screen
+        && now.duration_since(last_terminal_frame_at) >= ALT_SCREEN_IDLE_REPAINT_AFTER
+        && now.duration_since(last_attempt_at) >= ALT_SCREEN_IDLE_REPAINT_AFTER
+}
+
 fn arm_stale_terminal_input_suppression(suppress_until: &mut Option<Instant>) {
     *suppress_until = Some(Instant::now() + POST_RECONNECT_STALE_INPUT_GRACE);
 }
@@ -5679,19 +5804,21 @@ fn stale_mouse_report_maybe_incomplete(bytes: &[u8], offset: usize) -> bool {
         [0x1b] | [0x1b, b'['] | [0x1b, b'[', b'<'] => true,
         [0x1b, b'[', b'M', ..] if rest.len() < 6 => true,
         [0x1b, b'[', b'<', params @ ..] | [0x1b, b'[', params @ ..]
-            if params_are_mouse_prefix(params) =>
+            if params_are_mouse_prefix(params, 2) =>
         {
             true
         }
         [0x9b] | [0x9b, b'<'] => true,
         [0x9b, b'M', ..] if rest.len() < 5 => true,
-        [0x9b, b'<', params @ ..] | [0x9b, params @ ..] if params_are_mouse_prefix(params) => true,
+        [0x9b, b'<', params @ ..] | [0x9b, params @ ..] if params_are_mouse_prefix(params, 2) => {
-        params if params_are_mouse_prefix(params) => true,
+            true
+        }
+        params if params_are_mouse_prefix(params, 2) => true,
         _ => false,
     }
 }
 
-fn params_are_mouse_prefix(params: &[u8]) -> bool {
+fn params_are_mouse_prefix(params: &[u8], min_semicolons: usize) -> bool {
     let mut semicolons = 0usize;
     let mut saw_digit = false;
     for byte in params {
@@ -5701,7 +5828,7 @@ fn params_are_mouse_prefix(params: &[u8]) -> bool {
             _ => return false,
         }
     }
-    saw_digit && semicolons >= 1
+    saw_digit && semicolons >= min_semicolons
 }
 
 async fn flush_startup_input_if_ready(
@@ -7228,11 +7355,12 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
 #[cfg(test)]
 mod tests {
     use super::{
-        CachedCredential, DisconnectStatus, DynamicForward, FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer,
+        ALT_SCREEN_IDLE_REPAINT_AFTER, CachedCredential, DisconnectStatus, DynamicForward,
-        LocalForward, MAX_PENDING_USER_INPUT_BYTES, POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen,
+        FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer, LocalForward, MAX_PENDING_USER_INPUT_BYTES,
-        PredictMode, Predictor, RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD,
+        POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen, PredictMode, Predictor,
-        SshConfig, StartupGateMode, StatusAction, auth_allows, cache_key, cache_server_prefix,
+        RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD, SshConfig, StartupGateMode,
-        clear_cached_credentials, ensure_tui_safe_status_overlay, input_matches_escape,
+        StatusAction, auth_allows, cache_key, cache_server_prefix, clear_cached_credentials,
+        ensure_tui_safe_status_overlay, input_contains_focus_in, input_matches_escape,
         is_local_status_target, is_resume_response_for_client, latest_release_download_url,
         load_first_native_identity_with_prompt, parse_dynamic_forward, parse_escape_key,
         parse_local_forward, parse_remote_forward, parse_ssh_config, post_submit_hold_duration,
@@ -7241,9 +7369,10 @@ mod tests {
         render_status_clear, render_status_overlay, requested_env, resolved_startup_command,
         retransmit_stream_opens, rewrite_forward_command, selected_predict_mode, selected_udp_host,
         server_version_mismatch, should_flush_terminal_input_after_contact,
-        should_hold_during_startup_gate, should_hold_post_submit_input, split_after_command_submit,
+        should_hold_during_startup_gate, should_hold_post_submit_input,
-        ssh_destination_host, ssh_username, ssh_with_user, startup_command, status_ssh_target,
+        should_repaint_idle_alternate_screen, split_after_command_submit, ssh_destination_host,
-        strip_stale_mouse_reports, toml_bare_key_or_quoted, update_check_requested,
+        ssh_username, ssh_with_user, startup_command, status_ssh_target, strip_stale_mouse_reports,
+        strip_terminal_focus_reports, toml_bare_key_or_quoted, update_check_requested,
         update_version_status, upsert_managed_block, valid_forward_host, vscode_safe_alias,
     };
     use dosh::config::{ClientConfig, CommandExtension, HostConfig};
@@ -7253,7 +7382,7 @@ mod tests {
     use ed25519_dalek::{SigningKey, VerifyingKey};
     use std::collections::{HashMap, VecDeque};
     use std::fs;
-    use std::time::Duration;
+    use std::time::{Duration, Instant};
 
     fn test_args(server: &str, command: &[&str]) -> super::Args {
         super::Args {
@@ -8529,12 +8658,64 @@ mod tests {
         assert_eq!(strip_stale_mouse_reports(b"10m"), b"");
     }
 
+    #[test]
+    fn numeric_semicolon_input_survives_stale_filter() {
+        assert_eq!(strip_stale_mouse_reports(b"123;"), b"123;");
+        assert_eq!(strip_stale_mouse_reports(b"12;34"), b"12;34");
+        assert_eq!(strip_stale_mouse_reports(b"echo 1;"), b"echo 1;");
+    }
+
+    #[test]
+    fn incomplete_three_field_mouse_prefix_is_suppressed() {
+        assert_eq!(strip_stale_mouse_reports(b"35;152;"), b"");
+        assert_eq!(strip_stale_mouse_reports(b"\x1b[<35;152;"), b"");
+        assert_eq!(strip_stale_mouse_reports(b"\x9b35;152;"), b"");
+    }
+
     #[test]
     fn stale_focus_reports_are_stripped_from_pending_input() {
         assert_eq!(strip_stale_mouse_reports(b"\x1b[Ihello\x1b[O"), b"hello");
         assert_eq!(strip_stale_mouse_reports(b"\x9bIhello\x9bO"), b"hello");
     }
 
+    #[test]
+    fn focus_in_reports_trigger_local_repaint_detection() {
+        assert!(input_contains_focus_in(b"\x1b[I"));
+        assert!(input_contains_focus_in(b"before\x9bIafter"));
+        assert!(!input_contains_focus_in(b"\x1b[O"));
+        assert!(!input_contains_focus_in(b"\x1b[A"));
+    }
+
+    #[test]
+    fn focus_reports_are_never_forwarded_as_user_input() {
+        assert_eq!(strip_terminal_focus_reports(b"\x1b[Ihello\x1b[O"), b"hello");
+        assert_eq!(strip_terminal_focus_reports(b"\x9bIhello\x9bO"), b"hello");
+        assert_eq!(strip_terminal_focus_reports(b"\x1b[A"), b"\x1b[A");
+        assert_eq!(
+            strip_terminal_focus_reports(b"before\x1b[Iafter"),
+            b"beforeafter"
+        );
+    }
+
+    #[test]
+    fn idle_repaint_only_runs_for_stale_alternate_screen() {
+        let now = Instant::now();
+        let stale = now - ALT_SCREEN_IDLE_REPAINT_AFTER - Duration::from_secs(1);
+        let recent = now - Duration::from_secs(1);
+        assert!(should_repaint_idle_alternate_screen(
+            true, stale, stale, now
+        ));
+        assert!(!should_repaint_idle_alternate_screen(
+            false, stale, stale, now
+        ));
+        assert!(!should_repaint_idle_alternate_screen(
+            true, recent, stale, now
+        ));
+        assert!(!should_repaint_idle_alternate_screen(
+            true, stale, recent, now
+        ));
+    }
+
     #[test]
     fn non_mouse_escape_input_survives_stale_filter() {
         assert_eq!(strip_stale_mouse_reports(b"\x1b[A"), b"\x1b[A");

src/bin/dosh-server.rs

@@ -433,6 +433,7 @@ struct PendingStreamOpen {
     attempts: u32,
 }
 
+#[derive(Clone)]
 struct PendingNativeAuth {
     client: dosh::native::NativeClientHello,
     server: NativeServerHello,
@@ -864,6 +865,7 @@ async fn handle_packet(
         | PacketKind::StreamEof
         | PacketKind::StreamWindowAdjust
         | PacketKind::RekeyAck
+        | PacketKind::Detach
             if find_client_decrypt_key(state, &packet.header).is_err() =>
         {
             send_reject_to_client(socket, peer, packet.header.conn_id, "unknown client").await
@@ -992,8 +994,8 @@ async fn handle_native_user_auth(
     packet: &protocol::Packet,
 ) -> Result<()> {
     let pending = {
-        let mut locked = state.lock().expect("server state poisoned");
+        let locked = state.lock().expect("server state poisoned");
-        locked.pending_native.remove(&packet.header.conn_id)
+        locked.pending_native.get(&packet.header.conn_id).cloned()
     };
     let Some(pending) = pending else {
         return send_reject_to_client(socket, peer, packet.header.conn_id, "unknown native auth")
@@ -1009,11 +1011,38 @@ async fn handle_native_user_auth(
         .await;
     }
     if pending.created_at.elapsed() > Duration::from_secs(30) {
+        state
+            .lock()
+            .expect("server state poisoned")
+            .pending_native
+            .remove(&packet.header.conn_id);
         return send_reject_to_client(socket, peer, packet.header.conn_id, "native auth expired")
             .await;
     }
-    let body = protocol::decrypt_body(packet, &pending.session_key, CLIENT_TO_SERVER)?;
+    let body = match protocol::decrypt_body(packet, &pending.session_key, CLIENT_TO_SERVER) {
-    let req: NativeUserAuthBody = protocol::from_body(&body)?;
+        Ok(body) => body,
+        Err(_) => {
+            return send_reject_to_client(
+                socket,
+                peer,
+                packet.header.conn_id,
+                "native auth decrypt failed",
+            )
+            .await;
+        }
+    };
+    let req: NativeUserAuthBody = match protocol::from_body(&body) {
+        Ok(req) => req,
+        Err(_) => {
+            return send_reject_to_client(
+                socket,
+                peer,
+                packet.header.conn_id,
+                "invalid native auth body",
+            )
+            .await;
+        }
+    };
     if pending.client.requested_mode == "doctor" {
         let check_result = {
             let locked = state.lock().expect("server state poisoned");
@@ -1049,6 +1078,11 @@ async fn handle_native_user_auth(
                 .await;
             }
         };
+        state
+            .lock()
+            .expect("server state poisoned")
+            .pending_native
+            .remove(&packet.header.conn_id);
         let body = protocol::to_body(&check)?;
         let out = protocol::encode_encrypted(
             PacketKind::NativeAuthCheckOk,
@@ -1229,6 +1263,11 @@ async fn handle_native_user_auth(
     if let Some((listener, path)) = agent_listener {
         spawn_agent_forward(state, socket, client_id, listener, path);
     }
+    state
+        .lock()
+        .expect("server state poisoned")
+        .pending_native
+        .remove(&packet.header.conn_id);
 
     let ok = NativeAuthOk {
         client_id,
@@ -1696,7 +1735,14 @@ async fn handle_ping(
 }
 
 async fn handle_detach(state: &Arc<Mutex<ServerState>>, packet: &protocol::Packet) -> Result<()> {
+    let (key, _) = find_client_decrypt_key(state, &packet.header)?;
+    let _ = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
     let mut locked = state.lock().expect("server state poisoned");
+    if let Some(client) = locked.client_mut(&packet.header.conn_id) {
+        if !client.replay.accept(packet.header.seq) {
+            return Ok(());
+        }
+    }
     locked.remove_client_everywhere(&packet.header.conn_id);
     Ok(())
 }
@@ -4067,6 +4113,63 @@ mod tests {
         assert!(locked.lookup_client(&client_id).is_none());
     }
 
+    #[tokio::test]
+    async fn forged_plaintext_detach_does_not_remove_client() {
+        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
+        state
+            .ensure_session("work", 80, 24, "forward-only", &[])
+            .unwrap();
+        let client_id = [21u8; 16];
+        let session_key = [22u8; 32];
+        state.insert_client("work", client_id, test_client_state(session_key));
+        let state = Arc::new(Mutex::new(state));
+        let mut packet = protocol::decode(
+            &protocol::encode_plain(PacketKind::Detach, client_id, 1, 0, b"").unwrap(),
+        )
+        .unwrap();
+        packet.header.session_key_id = protocol::session_key_id(&session_key);
+
+        assert!(handle_detach(&state, &packet).await.is_err());
+
+        let locked = state.lock().unwrap();
+        assert!(
+            locked.lookup_client(&client_id).is_some(),
+            "plaintext detach with a copied key id must not remove a live client"
+        );
+    }
+
+    #[tokio::test]
+    async fn authenticated_detach_removes_client() {
+        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
+        state
+            .ensure_session("work", 80, 24, "forward-only", &[])
+            .unwrap();
+        let client_id = [23u8; 16];
+        let session_key = [24u8; 32];
+        state.insert_client("work", client_id, test_client_state(session_key));
+        let state = Arc::new(Mutex::new(state));
+        let packet = protocol::decode(
+            &protocol::encode_encrypted(
+                PacketKind::Detach,
+                client_id,
+                1,
+                0,
+                &session_key,
+                CLIENT_TO_SERVER,
+                b"",
+            )
+            .unwrap(),
+        )
+        .unwrap();
+
+        handle_detach(&state, &packet).await.unwrap();
+
+        let locked = state.lock().unwrap();
+        assert!(locked.lookup_client(&client_id).is_none());
+    }
+
     #[tokio::test]
     async fn duplicate_stream_open_for_open_stream_resends_ok() {
         let (pty_tx, _pty_rx) = mpsc::unbounded_channel();

src/protocol.rs

@@ -280,7 +280,7 @@ pub fn decode(input: &[u8]) -> Result<Packet> {
 
 pub fn decrypt_body(packet: &Packet, key: &[u8; 32], direction: u32) -> Result<Vec<u8>> {
     if packet.header.flags & 1 == 0 {
-        return Ok(packet.body.clone());
+        bail!("packet body is not encrypted");
     }
     let nonce = crypto::nonce_from(direction, packet.header.seq);
     let aad = packet.header.aad();

src/server.rs

@@ -13,6 +13,7 @@ use crate::transport::{
     DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
     service_name_from_target,
 };
+use crate::udp::is_transient_udp_send_error;
 use anyhow::{Context, Result, anyhow, bail};
 use ed25519_dalek::SigningKey;
 use std::collections::{HashMap, HashSet};
@@ -97,6 +98,7 @@ pub enum DoshServerEvent {
     Ignored,
 }
 
+#[derive(Debug, Clone)]
 struct PendingServerAuth {
     client: native::NativeClientHello,
     server: NativeServerHello,
@@ -255,7 +257,7 @@ impl DoshServer {
         };
         let body = protocol::to_body(&NativeServerHelloBody { hello })?;
         let out = protocol::encode_plain(PacketKind::NativeServerHello, pending_id, 1, 0, &body)?;
-        self.socket.send_to(&out, peer).await?;
+        let _ = send_udp(&self.socket, &out, peer).await?;
         Ok(())
     }
 
@@ -327,7 +329,7 @@ impl DoshServer {
         peer: SocketAddr,
         packet: &protocol::Packet,
     ) -> Result<DoshServerEvent> {
-        let Some(pending) = self.pending.remove(&packet.header.conn_id) else {
+        let Some(pending) = self.pending.get(&packet.header.conn_id).cloned() else {
             self.send_reject(peer, packet.header.conn_id, "unknown native auth")
                 .await?;
             return Ok(DoshServerEvent::Ignored);
@@ -338,13 +340,28 @@ impl DoshServer {
             return Ok(DoshServerEvent::Ignored);
         }
         if pending.created_at.elapsed() > self.config.auth_timeout {
+            self.pending.remove(&packet.header.conn_id);
             self.send_reject(peer, packet.header.conn_id, "native auth expired")
                 .await?;
             return Ok(DoshServerEvent::Ignored);
         }
 
-        let body = protocol::decrypt_body(packet, &pending.session_key, CLIENT_TO_SERVER)?;
+        let body = match protocol::decrypt_body(packet, &pending.session_key, CLIENT_TO_SERVER) {
-        let req: NativeUserAuthBody = protocol::from_body(&body)?;
+            Ok(body) => body,
+            Err(_) => {
+                self.send_reject(peer, packet.header.conn_id, "native auth decrypt failed")
+                    .await?;
+                return Ok(DoshServerEvent::Ignored);
+            }
+        };
+        let req: NativeUserAuthBody = match protocol::from_body(&body) {
+            Ok(req) => req,
+            Err(_) => {
+                self.send_reject(peer, packet.header.conn_id, "invalid native auth body")
+                    .await?;
+                return Ok(DoshServerEvent::Ignored);
+            }
+        };
         let services = match self.verify_auth_and_services(&pending, &req.auth) {
             Ok(services) => services,
             Err(err) => {
@@ -353,6 +370,7 @@ impl DoshServer {
                 return Ok(DoshServerEvent::Ignored);
             }
         };
+        self.pending.remove(&packet.header.conn_id);
 
         let conn_id = crypto::random_16();
         let key_id = protocol::session_key_id(&pending.session_key);
@@ -381,7 +399,7 @@ impl DoshServer {
             SERVER_TO_CLIENT,
             &body,
         )?;
-        self.socket.send_to(&out, peer).await?;
+        let _ = send_udp(&self.socket, &out, peer).await?;
 
         let transport = DoshTransport::new(
             Arc::clone(&self.socket),
@@ -428,7 +446,7 @@ impl DoshServer {
             reason: reason.to_string(),
         })?;
         let out = protocol::encode_plain(PacketKind::AttachReject, conn_id, 0, 0, &body)?;
-        self.socket.send_to(&out, peer).await?;
+        let _ = send_udp(&self.socket, &out, peer).await?;
         Ok(())
     }
 
@@ -439,6 +457,14 @@ impl DoshServer {
     }
 }
 
+async fn send_udp(socket: &UdpSocket, packet: &[u8], peer: SocketAddr) -> Result<bool> {
+    match socket.send_to(packet, peer).await {
+        Ok(_) => Ok(true),
+        Err(err) if is_transient_udp_send_error(&err) => Ok(false),
+        Err(err) => Err(err).with_context(|| format!("send UDP packet to {peer}")),
+    }
+}
+
 fn validate_requested_services(
     allowed_services: &HashSet<String>,
     requests: &[ForwardingRequest],
@@ -477,6 +503,7 @@ mod tests {
     use super::*;
     use crate::client::DoshClient;
     use crate::config::{ClientConfig, HostsConfig};
+    use crate::protocol::{CLIENT_TO_SERVER, NativeUserAuthBody};
     use crate::transport::TransportEvent;
     use ed25519_dalek::SigningKey;
 
@@ -588,4 +615,91 @@ mod tests {
             }
         }
     }
+
+    #[tokio::test]
+    async fn bad_native_auth_does_not_consume_pending_challenge() {
+        let dir = tempfile::tempdir().unwrap();
+        let host_key = dir.path().join("host_key");
+        let authorized_keys = dir.path().join("authorized_keys");
+        let signing = SigningKey::from_bytes(&[93u8; 32]);
+        let keypair = ssh_key::private::Ed25519Keypair::from(&signing);
+        let private =
+            ssh_key::PrivateKey::new(ssh_key::private::KeypairData::from(keypair), "").unwrap();
+        std::fs::write(
+            &authorized_keys,
+            format!("{}\n", private.public_key().to_openssh().unwrap()),
+        )
+        .unwrap();
+
+        let server_config = ServerConfig {
+            host_key: host_key.to_string_lossy().to_string(),
+            authorized_keys: vec![authorized_keys.to_string_lossy().to_string()],
+            ..ServerConfig::default()
+        };
+        let server_config = DoshServerConfig::new(server_config)
+            .bind_addr("127.0.0.1:0".parse().unwrap())
+            .require_current_user(false);
+        let mut server = DoshServer::bind(server_config).await.unwrap();
+        let peer: SocketAddr = "127.0.0.1:9".parse().unwrap();
+        let (client_secret, client_public) = native::generate_native_ephemeral();
+        let hello = native::NativeClientHello {
+            protocol_version: native::NATIVE_PROTOCOL_VERSION,
+            client_random: crypto::random_32(),
+            client_ephemeral_public: client_public,
+            requested_host: "127.0.0.1".to_string(),
+            requested_user: "sdk-user".to_string(),
+            requested_session: "test".to_string(),
+            requested_mode: "forward-only".to_string(),
+            terminal_size: (80, 24),
+            supported_aead: vec!["chacha20poly1305".to_string()],
+            supported_user_key_algorithms: vec!["ssh-ed25519".to_string()],
+            cached_host_key_fingerprint: None,
+            attach_ticket_envelope: None,
+            requested_env: Vec::new(),
+        };
+        let (pending_id, server_hello) = server.build_server_hello(hello.clone(), peer).unwrap();
+        let session_key = native::derive_native_session_key(
+            &client_secret,
+            server_hello.server_ephemeral_public,
+            &hello,
+            &server_hello,
+        )
+        .unwrap();
+        let auth = native::sign_user_auth(&signing, &hello, &server_hello, Vec::new()).unwrap();
+        let body = protocol::to_body(&NativeUserAuthBody { auth }).unwrap();
+        let bad = protocol::encode_encrypted(
+            PacketKind::NativeUserAuth,
+            pending_id,
+            2,
+            1,
+            &[7u8; 32],
+            CLIENT_TO_SERVER,
+            &body,
+        )
+        .unwrap();
+        let bad = protocol::decode(&bad).unwrap();
+
+        assert!(matches!(
+            server.handle_user_auth(peer, &bad).await.unwrap(),
+            DoshServerEvent::Ignored
+        ));
+        assert!(server.pending.contains_key(&pending_id));
+
+        let valid = protocol::encode_encrypted(
+            PacketKind::NativeUserAuth,
+            pending_id,
+            2,
+            1,
+            &session_key,
+            CLIENT_TO_SERVER,
+            &body,
+        )
+        .unwrap();
+        let valid = protocol::decode(&valid).unwrap();
+        assert!(matches!(
+            server.handle_user_auth(peer, &valid).await.unwrap(),
+            DoshServerEvent::Accepted(_)
+        ));
+        assert!(!server.pending.contains_key(&pending_id));
+    }
 }

src/transport.rs

@@ -737,11 +737,17 @@ impl DoshTransport {
             if packet.header.conn_id != self.conn_id {
                 continue;
             }
+            let plain = match protocol::decrypt_body(
+                &packet,
+                &self.session_key,
+                self.role.recv_direction(),
+            ) {
+                Ok(plain) => plain,
+                Err(_) => continue,
+            };
             if !self.replay.accept(packet.header.seq) {
                 continue;
             }
-            let plain =
-                protocol::decrypt_body(&packet, &self.session_key, self.role.recv_direction())?;
             self.peer_addr = peer;
             self.ack_seq = self.ack_seq.max(packet.header.seq);
             self.last_contact = Instant::now();
@@ -756,14 +762,21 @@ impl DoshTransport {
         datagram: &[u8],
         peer: SocketAddr,
     ) -> Result<SessionEvent> {
-        let packet = protocol::decode(datagram)?;
+        let packet = match protocol::decode(datagram) {
+            Ok(packet) => packet,
+            Err(_) => return Ok(SessionEvent::Ignored),
+        };
         if packet.header.conn_id != self.conn_id {
             return Ok(SessionEvent::Ignored);
         }
+        let plain =
+            match protocol::decrypt_body(&packet, &self.session_key, self.role.recv_direction()) {
+                Ok(plain) => plain,
+                Err(_) => return Ok(SessionEvent::Ignored),
+            };
         if !self.replay.accept(packet.header.seq) {
             return Ok(SessionEvent::Ignored);
         }
-        let plain = protocol::decrypt_body(&packet, &self.session_key, self.role.recv_direction())?;
         self.peer_addr = peer;
         self.ack_seq = self.ack_seq.max(packet.header.seq);
         self.last_contact = Instant::now();
@@ -1130,4 +1143,61 @@ mod tests {
         assert!(matches!(server.recv().await.unwrap(), SessionEvent::Ping));
         assert_eq!(server.peer_addr(), roaming_addr);
     }
+
+    #[tokio::test]
+    async fn unauthenticated_datagram_does_not_poison_replay_window() {
+        let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let peer: SocketAddr = "127.0.0.1:9".parse().unwrap();
+        let key = [11u8; 32];
+        let wrong_key = [12u8; 32];
+        let conn_id = [4u8; 16];
+        let mut transport = DoshTransport::new_owned(
+            socket,
+            SessionTransportConfig {
+                role: SessionRole::Client,
+                conn_id,
+                session_key: key,
+                peer_addr: peer,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+        let bad = protocol::encode_encrypted(
+            PacketKind::Pong,
+            conn_id,
+            9,
+            0,
+            &wrong_key,
+            SERVER_TO_CLIENT,
+            b"",
+        )
+        .unwrap();
+        let valid = protocol::encode_encrypted(
+            PacketKind::Pong,
+            conn_id,
+            9,
+            0,
+            &key,
+            SERVER_TO_CLIENT,
+            b"",
+        )
+        .unwrap();
+
+        assert!(matches!(
+            transport
+                .handle_datagram(b"not a dosh packet", peer)
+                .await
+                .unwrap(),
+            SessionEvent::Ignored
+        ));
+        assert!(matches!(
+            transport.handle_datagram(&bad, peer).await.unwrap(),
+            SessionEvent::Ignored
+        ));
+        assert!(matches!(
+            transport.handle_datagram(&valid, peer).await.unwrap(),
+            SessionEvent::Pong
+        ));
+    }
 }

tests/protocol_auth.rs

@@ -58,6 +58,19 @@ fn encrypted_packet_round_trips() {
     assert_eq!(plain, b"hello");
 }
 
+#[test]
+fn plaintext_packet_never_decrypts_as_authenticated_body() {
+    let key = crypto::random_32();
+    let mut decoded = protocol::decode(
+        &protocol::encode_plain(PacketKind::Input, crypto::random_16(), 1, 0, b"hello").unwrap(),
+    )
+    .unwrap();
+    decoded.header.session_key_id = protocol::session_key_id(&key);
+
+    let err = protocol::decrypt_body(&decoded, &key, CLIENT_TO_SERVER).unwrap_err();
+    assert!(err.to_string().contains("not encrypted"));
+}
+
 #[test]
 fn encrypted_packet_rejects_wrong_session_key_id_before_decrypt() {
     let key = crypto::random_32();