Harden SDK stream ids and duplicate opens

Cargo.lock

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

Cargo.toml

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

src/bin/dosh-client.rs

@@ -2498,11 +2498,15 @@ fn download_file(
         0
     };
     let mut file = if offset > 0 {
-        fs::OpenOptions::new()
+        let file = fs::OpenOptions::new()
             .read(true)
+            .write(true)
             .append(true)
             .open(local)
-            .with_context(|| format!("open {}", local.display()))?
+            .with_context(|| format!("open {}", local.display()))?;
+        file.set_len(offset)
+            .with_context(|| format!("truncate {}", local.display()))?;
+        file
     } else {
         fs::OpenOptions::new()
             .create(true)
@@ -6296,6 +6300,60 @@ const FLAG_TRIGGER_LOW_MS: f64 = 50.0;
 /// (covers a sudden latency spike on the very first slow keystroke).
 const GLITCH_FORCE_MS: u128 = 250;
 
+/// Enough recent bytes to join a CSI private-mode sequence split across frames
+/// without retaining arbitrary terminal output.
+const TERMINAL_OUTPUT_PARSE_TAIL: usize = 64;
+
+fn alternate_screen_after_output(mut active: bool, bytes: &[u8]) -> bool {
+    let mut offset = 0usize;
+    while offset < bytes.len() {
+        let Some((next, transition)) = terminal_private_mode_transition(bytes, offset) else {
+            offset += 1;
+            continue;
+        };
+        if let Some(enable) = transition {
+            active = enable;
+        }
+        offset = next.max(offset + 1);
+    }
+    active
+}
+
+fn terminal_private_mode_transition(bytes: &[u8], offset: usize) -> Option<(usize, Option<bool>)> {
+    let mut cursor = offset;
+    match bytes.get(cursor).copied()? {
+        0x1b if bytes.get(cursor + 1) == Some(&b'[') => cursor += 2,
+        0x9b => cursor += 1,
+        _ => return None,
+    }
+
+    if bytes.get(cursor) != Some(&b'?') {
+        return None;
+    }
+    cursor += 1;
+
+    let params_start = cursor;
+    while let Some(byte) = bytes.get(cursor).copied() {
+        match byte {
+            b'0'..=b'9' | b';' | b':' => cursor += 1,
+            b'h' | b'l' => {
+                let params = &bytes[params_start..cursor];
+                let touches_alt = terminal_private_params_include_alt_screen(params);
+                return Some((cursor + 1, touches_alt.then_some(byte == b'h')));
+            }
+            0x40..=0x7e => return Some((cursor + 1, None)),
+            _ => return None,
+        }
+    }
+    None
+}
+
+fn terminal_private_params_include_alt_screen(params: &[u8]) -> bool {
+    params
+        .split(|byte| matches!(byte, b';' | b':'))
+        .any(|param| matches!(param, b"47" | b"1047" | b"1049"))
+}
+
 /// One speculatively-echoed character on the current line.
 #[derive(Clone, Debug)]
 struct PredictedCell {
@@ -6313,6 +6371,9 @@ struct Predictor {
     /// as vim/htop); we never speculate there because we cannot model arbitrary
     /// cursor addressing safely.
     alternate_screen: bool,
+    /// Tail of recent terminal output, retained so alternate-screen transitions
+    /// split across UDP frames are still detected.
+    output_parse_tail: Vec<u8>,
 
     /// Predicted cells for the current line, left-to-right starting at the
     /// column where the cursor sat when this run of predictions began.
@@ -6359,6 +6420,7 @@ impl Predictor {
             mode,
             enabled: enabled && mode != PredictMode::Off,
             alternate_screen: false,
+            output_parse_tail: Vec::new(),
             cells: Vec::new(),
             cursor: 0,
             epoch: 1,
@@ -6381,6 +6443,7 @@ impl Predictor {
         self.epoch += 1;
         self.confirmed_epoch = self.epoch - 1;
         self.alternate_screen = false;
+        self.output_parse_tail.clear();
         self.oldest_pending_at = None;
     }
 
@@ -6537,20 +6600,20 @@ impl Predictor {
     /// and confirms outstanding predictions: a fresh frame means the server has
     /// re-rendered the line, so our speculative glyphs are superseded and erased.
     fn observe_output(&mut self, bytes: &[u8]) {
-        if contains_bytes(bytes, b"\x1b[?1049h")
+        let mut parse = Vec::with_capacity(self.output_parse_tail.len() + bytes.len());
-            || contains_bytes(bytes, b"\x1b[?1047h")
+        parse.extend_from_slice(&self.output_parse_tail);
-            || contains_bytes(bytes, b"\x1b[?47h")
+        parse.extend_from_slice(bytes);
-        {
+
-            self.alternate_screen = true;
+        let before = self.alternate_screen;
-            let _ = self.discard_all();
+        self.alternate_screen = alternate_screen_after_output(self.alternate_screen, &parse);
-        }
+        if self.alternate_screen != before {
-        if contains_bytes(bytes, b"\x1b[?1049l")
-            || contains_bytes(bytes, b"\x1b[?1047l")
-            || contains_bytes(bytes, b"\x1b[?47l")
-        {
-            self.alternate_screen = false;
             let _ = self.discard_all();
         }
+
+        self.output_parse_tail.clear();
+        let keep = parse.len().min(TERMINAL_OUTPUT_PARSE_TAIL);
+        self.output_parse_tail
+            .extend_from_slice(&parse[parse.len().saturating_sub(keep)..]);
     }
 
     /// Sample SRTT from a newly arrived frame and confirm/clear outstanding
@@ -7344,10 +7407,13 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
     "\x1b[?1002l",
     "\x1b[?1001l",
     "\x1b[?1000l",
+    "\x1b[?1004l",
     "\x1b[?1015l",
     "\x1b[?1006l",
     "\x1b[?1005l",
     "\x1b[?2004l",
+    "\x1b[?47l",
+    "\x1b[?1047l",
     "\x1b[?1049l"
 )
 .as_bytes();
@@ -7372,8 +7438,9 @@ mod tests {
         should_hold_during_startup_gate, should_hold_post_submit_input,
         should_repaint_idle_alternate_screen, split_after_command_submit, ssh_destination_host,
         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,
+        strip_terminal_focus_reports, terminal_private_mode_transition, toml_bare_key_or_quoted,
-        update_version_status, upsert_managed_block, valid_forward_host, vscode_safe_alias,
+        update_check_requested, update_version_status, upsert_managed_block, valid_forward_host,
+        vscode_safe_alias,
     };
     use dosh::config::{ClientConfig, CommandExtension, HostConfig};
     use dosh::native::EnvVar;
@@ -7892,6 +7959,61 @@ mod tests {
         assert!(!predictor.alternate_screen);
     }
 
+    #[test]
+    fn predictor_tracks_legacy_and_combined_alternate_screen_modes() {
+        let mut predictor = Predictor::new(true);
+
+        predictor.observe_output(b"\x1b[?1000;1006;1049h");
+        assert!(predictor.alternate_screen);
+
+        predictor.observe_output(b"\x1b[?1006;1000l");
+        assert!(
+            predictor.alternate_screen,
+            "leaving mouse modes must not imply leaving alternate screen"
+        );
+
+        predictor.observe_output(b"\x1b[?1047l");
+        assert!(!predictor.alternate_screen);
+
+        predictor.observe_output(b"\x1b[?47h");
+        assert!(predictor.alternate_screen);
+        predictor.observe_output(b"\x1b[?47l");
+        assert!(!predictor.alternate_screen);
+    }
+
+    #[test]
+    fn predictor_tracks_alternate_screen_split_across_frames() {
+        let mut predictor = Predictor::new(true);
+
+        predictor.observe_output(b"\x1b");
+        assert!(!predictor.alternate_screen);
+        predictor.observe_output(b"[?");
+        assert!(!predictor.alternate_screen);
+        predictor.observe_output(b"10");
+        assert!(!predictor.alternate_screen);
+        predictor.observe_output(b"49h");
+        assert!(predictor.alternate_screen);
+
+        predictor.observe_output(b"\x1b");
+        assert!(predictor.alternate_screen);
+        predictor.observe_output(b"[?104");
+        assert!(predictor.alternate_screen);
+        predictor.observe_output(b"9l");
+        assert!(!predictor.alternate_screen);
+    }
+
+    #[test]
+    fn private_mode_parser_ignores_non_alt_modes() {
+        assert_eq!(
+            terminal_private_mode_transition(b"\x1b[?1000;1006h", 0),
+            Some((13, None))
+        );
+        assert_eq!(
+            terminal_private_mode_transition(b"\x1b[?1000;1049h", 0),
+            Some((13, Some(true)))
+        );
+    }
+
     /// Typing printable characters builds an in-order predicted line and the
     /// local cursor tracks the end of it.
     #[test]

src/bin/dosh-server.rs

@@ -2667,10 +2667,13 @@ async fn handle_file_request(
             let (file, temp_path, written, atomic) = if resume && final_path.exists() {
                 let mut file = fs::OpenOptions::new()
                     .read(true)
+                    .write(true)
                     .append(true)
                     .open(&final_path)
                     .with_context(|| format!("open {}", final_path.display()))?;
                 let written = file.metadata()?.len().min(size);
+                file.set_len(written)
+                    .with_context(|| format!("truncate {}", final_path.display()))?;
                 if written > 0 {
                     let mut prefix = fs::File::open(&final_path)?;
                     hash_exact_prefix(&mut prefix, written, &mut hasher)?;

src/transport.rs

@@ -121,6 +121,10 @@ pub struct AcceptedService {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum TransportEvent {
     Open(IncomingStreamOpen),
+    DuplicateOpen {
+        stream_id: u64,
+        response: OutgoingStreamPacket,
+    },
     OpenOk {
         stream_id: u64,
         flushed: Vec<OutgoingStreamPacket>,
@@ -258,6 +262,12 @@ impl StreamMux {
     }
 
     pub fn accept_open(&mut self, open: StreamOpen) -> Result<IncomingStreamOpen> {
+        if self.pending_opens.contains_key(&open.stream_id) {
+            bail!(
+                "stream {} collides with a local pending open",
+                open.stream_id
+            );
+        }
         self.opened_streams.insert(open.stream_id);
         self.send_credit
             .entry(open.stream_id)
@@ -309,8 +319,17 @@ impl StreamMux {
         if bytes.is_empty() {
             return Ok(Vec::new());
         }
-        if !self.opened_streams.contains(&stream_id)
+        if !self.opened_streams.contains(&stream_id) {
-            || self.send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len()
+            if self.pending_opens.contains_key(&stream_id) {
+                self.pending_data
+                    .entry(stream_id)
+                    .or_default()
+                    .push_back(bytes);
+                return Ok(Vec::new());
+            }
+            bail!("stream {stream_id} is not open");
+        }
+        if self.send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len()
             || self
                 .pending_data
                 .get(&stream_id)
@@ -367,6 +386,12 @@ impl StreamMux {
         match kind {
             PacketKind::StreamOpen => {
                 let open: StreamOpen = protocol::from_body(body)?;
+                if self.opened_streams.contains(&open.stream_id) {
+                    return Ok(TransportEvent::DuplicateOpen {
+                        stream_id: open.stream_id,
+                        response: Self::open_ok(open.stream_id)?,
+                    });
+                }
                 self.accept_open(open).map(TransportEvent::Open)
             }
             PacketKind::StreamOpenOk => {
@@ -625,7 +650,7 @@ impl DoshTransport {
             ack_seq: config.initial_ack,
             replay: ReplayWindow::default(),
             mux: StreamMux::new(config.stream),
-            next_stream_id: 1,
+            next_stream_id: initial_stream_id(config.role),
             last_contact: now,
             last_keepalive: now,
         }
@@ -669,7 +694,10 @@ impl DoshTransport {
 
     pub fn allocate_stream_id(&mut self) -> u64 {
         let id = self.next_stream_id;
-        self.next_stream_id = self.next_stream_id.saturating_add(1).max(1);
+        self.next_stream_id = self
+            .next_stream_id
+            .wrapping_add(1)
+            .max(initial_stream_id(self.role));
         id
     }
 
@@ -811,6 +839,9 @@ impl DoshTransport {
 
     async fn send_followups(&mut self, event: &TransportEvent) -> Result<()> {
         match event {
+            TransportEvent::DuplicateOpen { response, .. } => {
+                self.send_outgoing(response.clone()).await?;
+            }
             TransportEvent::OpenOk { flushed, .. }
             | TransportEvent::WindowAdjust { flushed, .. } => {
                 for packet in flushed.clone() {
@@ -857,6 +888,13 @@ impl DoshTransport {
     }
 }
 
+fn initial_stream_id(role: SessionRole) -> u64 {
+    match role {
+        SessionRole::Client => 1,
+        SessionRole::Server => 1u64 << 63,
+    }
+}
+
 pub fn service_target(name: &str) -> Result<String> {
     if name.is_empty()
         || !name
@@ -1047,6 +1085,152 @@ mod tests {
         );
     }
 
+    #[test]
+    fn duplicate_stream_open_reacks_without_second_open_event() {
+        let mut mux = StreamMux::new(TransportConfig::default());
+        let open = StreamOpen {
+            stream_id: 9,
+            target_host: "@dosh-app".to_string(),
+            target_port: 0,
+        };
+        let body = protocol::to_body(&open).unwrap();
+        assert!(matches!(
+            mux.handle_packet(PacketKind::StreamOpen, &body).unwrap(),
+            TransportEvent::Open(_)
+        ));
+
+        let duplicate = mux.handle_packet(PacketKind::StreamOpen, &body).unwrap();
+        match duplicate {
+            TransportEvent::DuplicateOpen {
+                stream_id,
+                response,
+            } => {
+                assert_eq!(stream_id, 9);
+                assert_eq!(response.kind, PacketKind::StreamOpenOk);
+                let ok: StreamOpenOk = decode(&response);
+                assert_eq!(ok.stream_id, 9);
+            }
+            other => panic!("unexpected duplicate open event {other:?}"),
+        }
+    }
+
+    #[test]
+    fn send_data_to_closed_or_unknown_stream_fails() {
+        let mut mux = StreamMux::new(TransportConfig::default());
+        mux.open_stream(1, "@dosh-test", 0).unwrap();
+        assert!(mux.send_data(1, b"queued".to_vec()).unwrap().is_empty());
+        mux.handle_open_ok(StreamOpenOk { stream_id: 1 }).unwrap();
+        mux.close_stream(1).unwrap();
+
+        assert!(mux.send_data(1, b"lost".to_vec()).is_err());
+        assert!(mux.send_data(99, b"lost".to_vec()).is_err());
+    }
+
+    #[tokio::test]
+    async fn session_transport_partitions_client_and_server_stream_ids() {
+        let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let client_addr = client_socket.local_addr().unwrap();
+        let server_addr = server_socket.local_addr().unwrap();
+        let key = [44u8; 32];
+        let conn_id = [8u8; 16];
+        let mut client = DoshTransport::new_owned(
+            client_socket,
+            SessionTransportConfig {
+                role: SessionRole::Client,
+                conn_id,
+                session_key: key,
+                peer_addr: server_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+        let mut server = DoshTransport::new_owned(
+            server_socket,
+            SessionTransportConfig {
+                role: SessionRole::Server,
+                conn_id,
+                session_key: key,
+                peer_addr: client_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+
+        assert_eq!(client.allocate_stream_id(), 1);
+        assert_eq!(client.allocate_stream_id(), 2);
+        assert_eq!(server.allocate_stream_id(), 1u64 << 63);
+        assert_eq!(server.allocate_stream_id(), (1u64 << 63) + 1);
+    }
+
+    #[tokio::test]
+    async fn session_transport_handles_simultaneous_opens_without_id_collision() {
+        let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let client_addr = client_socket.local_addr().unwrap();
+        let server_addr = server_socket.local_addr().unwrap();
+        let key = [45u8; 32];
+        let conn_id = [9u8; 16];
+        let mut client = DoshTransport::new_owned(
+            client_socket,
+            SessionTransportConfig {
+                role: SessionRole::Client,
+                conn_id,
+                session_key: key,
+                peer_addr: server_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+        let mut server = DoshTransport::new_owned(
+            server_socket,
+            SessionTransportConfig {
+                role: SessionRole::Server,
+                conn_id,
+                session_key: key,
+                peer_addr: client_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+
+        let client_stream = client.open_service("client_app").await.unwrap();
+        let server_stream = server.open_service("server_app").await.unwrap();
+        assert_ne!(client_stream, server_stream);
+
+        let server_seen = server.recv().await.unwrap();
+        let client_seen = client.recv().await.unwrap();
+        match server_seen {
+            SessionEvent::Stream(TransportEvent::Open(open)) => {
+                assert_eq!(open.stream_id, client_stream);
+                assert_eq!(open.target_host, "@dosh-client_app");
+                server.accept_stream(open.stream_id).await.unwrap();
+            }
+            other => panic!("unexpected server event {other:?}"),
+        }
+        match client_seen {
+            SessionEvent::Stream(TransportEvent::Open(open)) => {
+                assert_eq!(open.stream_id, server_stream);
+                assert_eq!(open.target_host, "@dosh-server_app");
+                client.accept_stream(open.stream_id).await.unwrap();
+            }
+            other => panic!("unexpected client event {other:?}"),
+        }
+
+        assert!(matches!(
+            client.recv().await.unwrap(),
+            SessionEvent::Stream(TransportEvent::OpenOk { stream_id, .. }) if stream_id == client_stream
+        ));
+        assert!(matches!(
+            server.recv().await.unwrap(),
+            SessionEvent::Stream(TransportEvent::OpenOk { stream_id, .. }) if stream_id == server_stream
+        ));
+    }
+
     #[tokio::test]
     async fn session_transport_opens_service_and_moves_data_over_encrypted_udp() {
         let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();

tests/integration_smoke.rs

@@ -1108,6 +1108,117 @@ fn native_file_copy_recursive_round_trip() {
     let _ = server.wait();
 }
 
+#[test]
+fn native_file_copy_resume_truncates_oversized_destinations() {
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config(&dir, port);
+    write_native_client_auth(&dir, &config);
+    let mut server = start_server(&dir, &config);
+    let client_bin = env!("CARGO_BIN_EXE_dosh-client");
+
+    let local_short = dir.path().join("short.txt");
+    fs::write(&local_short, b"short\n").unwrap();
+
+    let seed_remote = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("exec")
+        .arg("local")
+        .arg("printf 'short\\nSTALE' > \"$HOME/resume-upload.txt\"")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        seed_remote.status.success(),
+        "seed remote failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&seed_remote.stdout),
+        String::from_utf8_lossy(&seed_remote.stderr)
+    );
+
+    let resume_upload = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("--resume")
+        .arg(&local_short)
+        .arg("local:resume-upload.txt")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        resume_upload.status.success(),
+        "resume upload failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&resume_upload.stdout),
+        String::from_utf8_lossy(&resume_upload.stderr)
+    );
+
+    let cat_upload = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cat")
+        .arg("local:resume-upload.txt")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        cat_upload.status.success(),
+        "cat upload failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&cat_upload.stdout),
+        String::from_utf8_lossy(&cat_upload.stderr)
+    );
+    assert_eq!(String::from_utf8_lossy(&cat_upload.stdout), "short\n");
+
+    let seed_download = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("exec")
+        .arg("local")
+        .arg("printf 'tiny\\n' > \"$HOME/resume-download.txt\"")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        seed_download.status.success(),
+        "seed download failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&seed_download.stdout),
+        String::from_utf8_lossy(&seed_download.stderr)
+    );
+
+    let local_download = dir.path().join("resume-download-local.txt");
+    fs::write(&local_download, b"tiny\nSTALE").unwrap();
+    let resume_download = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("--resume")
+        .arg("local:resume-download.txt")
+        .arg(&local_download)
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        resume_download.status.success(),
+        "resume download failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&resume_download.stdout),
+        String::from_utf8_lossy(&resume_download.stderr)
+    );
+    assert_eq!(fs::read_to_string(&local_download).unwrap(), "tiny\n");
+
+    let _ = server.kill();
+    let _ = server.wait();
+}
+
 #[test]
 fn native_exec_command_smoke() {
     let dir = tempfile::tempdir().unwrap();