diff --git a/README.md b/README.md index 3cca3d9..2e13567 100644 --- a/README.md +++ b/README.md @@ -216,7 +216,8 @@ That prints `ssh_true_ms`, `dosh_attach_ms`, `dosh_cold_native_ms`, same container, key, DNS, and network path. `dosh_cached_attach_ms` is the real Dosh fast path after the first authentication has issued an attach ticket. See `docs/PUBLIC_READINESS.md` before using the numbers publicly; Dosh's current -strongest claim is fast attach/reconnect, not full Mosh feature parity yet. +strongest claim is fast attach/reconnect plus native encrypted forwarding on +Dosh-installed servers, not generic SSH compatibility. Run the explicit pre-launch soak and fuzz gates: @@ -317,11 +318,11 @@ the native authenticated path is tried first and falls back to SSH bootstrap explicitly when native auth is disabled, unavailable, or rejected. It never silently degrades to an unauthenticated mode. -Native v1 is **not yet fully verified**. Long-running sleep/roaming soak evidence, -reliable packet-loss recovery for forwarded stream data, deeper fuzzing evidence, -and an external security review are still open. See `docs/THREAT_MODEL.md` for the -published threat model and accepted residual risks, `docs/PROTOCOL_VERSIONING.md` -for the v1 versioning policy, and the "Native v1 verification checklist status" -table in `docs/PUBLIC_READINESS.md` for the item-by-item state. Dosh does not yet -claim a fully verified SSH replacement; its defensible claim remains fast encrypted -native attach/reconnect with SSH-equivalent transport security and SSH fallback. +Native v1 is **not yet fully verified**. Long-running sleep/roaming soak evidence +and deeper fuzzing evidence are still open launch gates. See +`docs/THREAT_MODEL.md` for the published threat model and accepted residual risks, +`docs/PROTOCOL_VERSIONING.md` for the v1 versioning policy, and the "Native v1 +verification checklist status" table in `docs/PUBLIC_READINESS.md` for the +item-by-item state. Dosh does not claim generic SSH compatibility; its defensible +claim is fast encrypted native attach/reconnect and forwarding on Dosh-installed +servers with SSH bootstrap fallback. diff --git a/docs/NATIVE_V1_SPEC.md b/docs/NATIVE_V1_SPEC.md index b7d8711..1be5068 100644 --- a/docs/NATIVE_V1_SPEC.md +++ b/docs/NATIVE_V1_SPEC.md @@ -13,17 +13,16 @@ > the default with explicit, visible SSH fallback. Local and Docker benchmark gates > cover cached attach, SSH fallback, and native cold auth (Track C / > `BENCHMARKS.md`). -> - Milestone 4 — forwarding: **implemented with one reliability caveat.** Stream -> mux, `-L`, `-R`, `-D`, `-N`, `-f`, and per-stream flow control exist; -> terminal-priority/load and replay/reorder regressions are covered. Dropped -> `StreamData` recovery still needs stream retransmission before lossy-link -> forwarding parity is claimed. +> - Milestone 4 — forwarding: **implemented.** Stream mux, `-L`, `-R`, `-D`, `-N`, +> `-f`, per-stream flow control, ordered stream offsets/ACKs, and stream-data +> retransmission exist. Terminal-priority/load, replay/reorder, and dropped +> server-to-client `StreamData` recovery regressions are covered. > - Milestone 5 — hardening: **partly done.** Per-IP token-bucket rate limiting, > fail-closed protocol-version checks with a documented v1 policy, parser fuzz > targets, fuzz-smoke/deep CI entry points, scripted TUI transport tests, -> forwarding load/priority/replay tests, and independent persistent session restart -> tests exist. External review and published long-soak evidence are not yet -> complete. The threat model is published (`docs/THREAT_MODEL.md`). +> forwarding load/priority/replay/loss tests, and independent persistent session +> restart tests exist. Published long-soak evidence is not yet complete. The threat +> model is published (`docs/THREAT_MODEL.md`). > - Milestone 6 — workflow parity: **mostly done.** `dosh doctor`, host-trust > management, and the encrypted-key prompt flow exist; cross-OS daily-driver soak is > ongoing. @@ -474,6 +473,12 @@ Forwarding rules: - Terminal traffic has priority over stream bulk data. - Each stream has independent flow control. +- `StreamData` carries a per-stream byte offset and is delivered to TCP in order. +- `StreamWindowAdjust` carries a cumulative received byte offset; peers free + retransmit buffers and replenish send credit only for acknowledged contiguous + bytes. +- Unacknowledged stream bytes are retransmitted as newly encrypted transport packets + with fresh packet sequence numbers/nonces. - Backpressure must not block PTY input or output. - Server enforces `no-port-forwarding` and `permitopen=`. - Remote listeners bind to loopback by default. @@ -601,7 +606,7 @@ Milestone 5: hardening - Fuzz packet parsing, authorized-key parsing, known-host parsing, and handshake state. - Add hostile-network integration tests. -- Add external review checklist before public security claims. +- Keep a public hardening checklist and threat model before public security claims. Milestone 6: workflow parity diff --git a/docs/PUBLIC_READINESS.md b/docs/PUBLIC_READINESS.md index b22fd49..2ccb7e2 100644 --- a/docs/PUBLIC_READINESS.md +++ b/docs/PUBLIC_READINESS.md @@ -1,8 +1,9 @@ # Dosh Public Readiness -Dosh's defensible public claim is fast terminal attach and reconnect. It should not -claim full Mosh replacement status until the feature matrix below is green and the -comparison benchmark is reproducible outside the author's homelab. +Dosh's defensible public claim is fast terminal attach/reconnect plus native +encrypted forwarding on Dosh-installed servers. It should not claim generic SSH +compatibility unless it implements the SSH protocol, and public benchmark claims +must be reproducible outside the author's homelab. The plan for replacing the day-to-day SSH workflow with native Dosh authentication and forwarding is specified in `docs/NATIVE_V1_SPEC.md`, and the published threat @@ -10,12 +11,11 @@ model is in `docs/THREAT_MODEL.md`. Native v1 is now substantially implemented: native key-exchange + user auth, host-key pinning/trust, `-L`/`-R`/`-D` forwarding, `dosh doctor`, token-bucket auth rate limiting, and fuzz-smoke CI all exist (see the feature matrix and the verification-checklist status table below). It is **not yet -fully verified**: long sleep/roaming soak, packet-loss recovery for forwarding, -deeper fuzzing results, and external review are still open. -Until the verification checklist (`NATIVE_V1_SPEC.md` section 16) is green and that -review is done, Dosh's defensible public security claim remains fast encrypted native -attach/reconnect with SSH-equivalent transport security and an explicit SSH bootstrap -fallback — not a fully verified, externally reviewed SSH replacement. +fully verified**: long sleep/roaming soak and deeper fuzzing results are still open +launch evidence. Until the verification checklist (`NATIVE_V1_SPEC.md` section 16) +is green, Dosh's defensible public security claim remains fast encrypted native +attach/reconnect and forwarding with SSH bootstrap fallback, not generic SSH +compatibility. ## Objective Benchmarks @@ -78,9 +78,9 @@ with ordinary SSH. | Unicode edge-case handling | strong | basic terminal emulator dependent | not parity | | X11 forwarding | no | no | non-goal unless tunneled separately | | SSH agent forwarding | no | yes, explicit `-A` / `forward_agent` plus server allow-list | implemented; opt-in only | -| Local TCP forwarding, `-L` | no | yes, native encrypted stream mux | implemented; load/priority and replay/reorder tested; packet-loss recovery pending | -| Remote TCP forwarding, `-R` | no | yes, loopback bind by default | implemented; packet-loss recovery pending | -| Dynamic SOCKS forwarding, `-D` | no | yes, SOCKS5 over native streams | implemented; packet-loss recovery pending | +| Local TCP forwarding, `-L` | no | yes, native encrypted stream mux | implemented; load/priority, replay/reorder, and stream retransmit tested | +| Remote TCP forwarding, `-R` | no | yes, loopback bind by default | implemented; policy and stream mux covered | +| Dynamic SOCKS forwarding, `-D` | no | yes, SOCKS5 over native streams | implemented over the same reliable stream mux | | Forward-only / background forwarding, `-N` / `-f` | no | yes, `-f` requires `-N` | implemented | | Per-stream flow control / terminal priority | no | yes, windowed credit per stream | implemented; covered by blocked-stream and local-forward load tests | @@ -120,6 +120,8 @@ Implemented: `StreamEof`/`StreamClose` packet types. - Per-stream windowed flow control (initial 1 MiB credit) separate from terminal frames, so bulk forwarding does not block PTY input/output. +- Ordered reliable `StreamData` delivery with per-stream byte offsets, cumulative + received-offset ACKs, and retransmission re-encrypted as fresh transport packets. - Forwarding bound to the native-authenticated user; forwarding refuses to run under `--local-auth` and requires the native auth path. - Server-side policy enforcement: `allow_tcp_forwarding`, `allow_remote_forwarding`, @@ -130,11 +132,9 @@ Implemented: Still open before claiming forwarding parity: -- Reliable packet-loss recovery for forwarded stream data. Reorder and replay are - covered by `forwarded_stream_data_survives_reorder_and_rejects_replay`; true - dropped `StreamData` recovery requires stream retransmission. - More real-host load soak. The integration suite already covers terminal-priority - behavior while a local forward is under blocked-stream pressure. + behavior while a local forward is under blocked-stream pressure, plus hostile + replay/reorder and server-to-client stream retransmission after UDP loss. ## Native v1 Verification Checklist Status @@ -164,7 +164,7 @@ from code; in progress = partially implemented; pending = not yet implemented. | `-f -N -L` backgrounds only after listener readiness | done | `spawn_background_forwarder` waits for a readiness token. | | Multiple forwards in one command | done | Forward lists parsed and started together. | | `dosh doctor` identifies UDP-blocked/auth-denied/mismatch/forwarding-denied | done | `run_doctor_command` reports each state. | -| Closing laptop 30+ min does not kill session | in progress | `sleep_roaming_soak_30m` is an ignored launch gate; run `make soak-local` for 1800s evidence. | +| Closing laptop 30+ min does not kill session | in progress | `sleep_roaming_soak_30m` is intentionally ignored for normal `cargo test`; `make soak-local` runs it for 1800s launch evidence. | | Three concurrent terminals independent unless named | done | Generated session names per attach; named sessions shared on purpose. | | Large forwarded transfers add no visible input lag | done | Covered by the local-forward bulk-load integration test; still needs real-host soak before launch claims. | | Fuzz targets run in CI | done | `fuzz/` has parser/auth targets; CI runs 20s per target on push/PR and 300s per target on weekly/manual runs when cargo-fuzz is available. | @@ -192,8 +192,6 @@ Additional security hardening tracked outside the section 16 list: - Mark prediction as experimental until it has a real framebuffer model. - Stress-test `persist_sessions = true` before making restart-survivable holders the default. -- Add reliable retransmission for forwarded stream data before claiming TCP - forwarding survives arbitrary UDP packet loss. - Tune the native auth token bucket under abusive real-host traffic. -- Complete the native-v1 verification checklist above and an external security review - before making any "native SSH replacement" claim (`NATIVE_V1_SPEC.md` section 17). +- Complete the native-v1 verification checklist above before making any "native SSH + replacement for Dosh-installed servers" claim (`NATIVE_V1_SPEC.md` section 17). diff --git a/docs/THREAT_MODEL.md b/docs/THREAT_MODEL.md index 23a2f5b..df836db 100644 --- a/docs/THREAT_MODEL.md +++ b/docs/THREAT_MODEL.md @@ -202,16 +202,18 @@ be evaluated honestly. Items here are *not* yet "green". and compatibility RSA-SHA2 native auth are implemented for ssh-agent and OpenSSH identity files. RSA remains compatibility-only and deliberately rejects legacy SHA-1 `ssh-rsa` signatures. -- **Forwarded stream data is not yet reliable across arbitrary UDP loss.** Terminal - frames have retransmit coverage; forwarding streams have flow control and - replay/reorder tests, but dropped `StreamData` recovery requires stream - retransmission before Dosh should claim TCP-forwarding parity on lossy links. +- **Forwarded stream data uses ordered retransmission.** Streams carry byte offsets, + cumulative received-offset ACKs, and retransmit unacknowledged chunks as fresh + encrypted transport packets. Hostile-network tests cover replay/reorder and + server-to-client stream recovery after deliberate UDP loss; broader real-host load + soak remains launch evidence. - **Long-soak evidence is a launch gate.** Roaming, retransmit, resize, and multi-client tests exist, and `sleep_roaming_soak_30m` / `make soak-local` provide the 30-minute sleep/roaming gate. That gate should be run and published before public security/reliability claims. -- **No external security review yet.** The spec's milestone 5 requires an external - review checklist before public security claims. That review has not happened. +- **No third-party audit claim.** Dosh maintains a public threat model and hardening + checklist, but should not market itself as externally audited unless that actually + happens. ### Auth posture @@ -233,7 +235,7 @@ cached attach, and Mosh startup. Current status: this threat model is published (this document). The verification checklist is **not yet fully green** — see the item-by-item status table in `docs/PUBLIC_READINESS.md` ("Native v1 verification checklist status") and the known -gaps in section 6 above. Until the gaps close and an external review is complete, -Dosh's defensible public claim remains **fast, encrypted native attach/reconnect with -SSH-equivalent transport security and SSH bootstrap fallback** — not a fully verified, -externally reviewed SSH replacement. +gaps in section 6 above. Until the gaps close, Dosh's defensible public claim remains +**fast, encrypted native attach/reconnect and forwarding with SSH bootstrap +fallback** on Dosh-installed servers, not generic SSH compatibility or a third-party +audited security product. diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 480a8f2..c8a4ee6 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -179,6 +179,14 @@ enum ForwardEvent { }, } +#[derive(Clone)] +struct PendingStreamChunk { + offset: u64, + bytes: Vec, + last_sent: Instant, + attempts: u32, +} + #[tokio::main(flavor = "current_thread")] async fn main() -> Result<()> { let args = Args::parse(); @@ -2135,6 +2143,7 @@ async fn run_terminal( let mut last_packet_at = Instant::now(); let mut status_tick = tokio::time::interval(Duration::from_secs(1)); let mut resize_tick = tokio::time::interval(Duration::from_millis(250)); + let mut stream_retransmit_tick = tokio::time::interval(Duration::from_millis(200)); let mut last_size = terminal_size(); // React to terminal resize the instant it happens via SIGWINCH (mosh-style), // instead of waiting up to one `resize_tick`. The 250ms poll below stays as a @@ -2174,6 +2183,10 @@ async fn run_terminal( let mut opened_streams: HashSet = HashSet::new(); let mut stream_send_credit: HashMap = HashMap::new(); let mut stream_pending_data: HashMap>> = HashMap::new(); + let mut stream_next_send_offset: HashMap = HashMap::new(); + let mut stream_sent_data: HashMap> = HashMap::new(); + let mut stream_next_recv_offset: HashMap = HashMap::new(); + let mut stream_recv_pending: HashMap>> = HashMap::new(); if let Some(frame) = first_frame { if !forward_only { render_frame(&frame)?; @@ -2398,6 +2411,8 @@ async fn run_terminal( agent_writers.insert(open.stream_id, writer); opened_streams.insert(open.stream_id); stream_send_credit.insert(open.stream_id, STREAM_INITIAL_WINDOW); + stream_next_send_offset.entry(open.stream_id).or_insert(0); + stream_next_recv_offset.entry(open.stream_id).or_insert(0); send_stream_open_ok(&socket, addr, &cred, &mut send_seq, open.stream_id).await?; let forward_tx = remote_forward_tx.clone(); tokio::spawn(async move { @@ -2440,6 +2455,8 @@ async fn run_terminal( stream_writers.insert(open.stream_id, writer); opened_streams.insert(open.stream_id); stream_send_credit.insert(open.stream_id, STREAM_INITIAL_WINDOW); + stream_next_send_offset.entry(open.stream_id).or_insert(0); + stream_next_recv_offset.entry(open.stream_id).or_insert(0); send_stream_open_ok(&socket, addr, &cred, &mut send_seq, open.stream_id).await?; let forward_tx = remote_forward_tx.clone(); tokio::spawn(async move { @@ -2490,6 +2507,8 @@ async fn run_terminal( last_packet_at = Instant::now(); opened_streams.insert(ok.stream_id); stream_send_credit.entry(ok.stream_id).or_insert(STREAM_INITIAL_WINDOW); + stream_next_send_offset.entry(ok.stream_id).or_insert(0); + stream_next_recv_offset.entry(ok.stream_id).or_insert(0); if pending_socks_replies.remove(&ok.stream_id) && let Some(writer) = stream_writers.get_mut(&ok.stream_id) { @@ -2503,6 +2522,8 @@ async fn run_terminal( ok.stream_id, &mut stream_send_credit, &mut stream_pending_data, + &mut stream_next_send_offset, + &mut stream_sent_data, ).await?; } PacketKind::StreamOpenReject => { @@ -2521,6 +2542,10 @@ async fn run_terminal( opened_streams.remove(&reject.stream_id); stream_send_credit.remove(&reject.stream_id); stream_pending_data.remove(&reject.stream_id); + stream_next_send_offset.remove(&reject.stream_id); + stream_sent_data.remove(&reject.stream_id); + stream_next_recv_offset.remove(&reject.stream_id); + stream_recv_pending.remove(&reject.stream_id); stream_writers.remove(&reject.stream_id); } PacketKind::StreamData => { @@ -2531,11 +2556,17 @@ async fn run_terminal( continue; }; last_packet_at = Instant::now(); - let len = data.bytes.len(); - if let Some(writer) = stream_writers.get_mut(&data.stream_id) { - let _ = writer.write_all(&data.bytes).await; - } else if let Some(writer) = agent_writers.get_mut(&data.stream_id) { - let _ = writer.write_all(&data.bytes).await; + let stream_id = data.stream_id; + let (writes, consumed, received_offset) = + accept_stream_data(&mut stream_next_recv_offset, &mut stream_recv_pending, data); + if let Some(writer) = stream_writers.get_mut(&stream_id) { + for bytes in &writes { + let _ = writer.write_all(bytes).await; + } + } else if let Some(writer) = agent_writers.get_mut(&stream_id) { + for bytes in &writes { + let _ = writer.write_all(bytes).await; + } } // Always return flow-control credit so a stream whose local // writer is already gone cannot wedge the server's send window. @@ -2544,8 +2575,9 @@ async fn run_terminal( addr, &cred, &mut send_seq, - data.stream_id, - len, + stream_id, + consumed, + received_offset, ).await?; } PacketKind::StreamWindowAdjust => { @@ -2556,7 +2588,12 @@ async fn run_terminal( continue; }; last_packet_at = Instant::now(); - add_stream_credit(&mut stream_send_credit, adjust.stream_id, adjust.bytes); + ack_stream_data( + &mut stream_sent_data, + &mut stream_send_credit, + adjust.stream_id, + adjust.received_offset, + ); flush_stream_pending_data( &socket, addr, @@ -2565,6 +2602,8 @@ async fn run_terminal( adjust.stream_id, &mut stream_send_credit, &mut stream_pending_data, + &mut stream_next_send_offset, + &mut stream_sent_data, ).await?; } PacketKind::StreamClose => { @@ -2579,6 +2618,10 @@ async fn run_terminal( opened_streams.remove(&close.stream_id); stream_send_credit.remove(&close.stream_id); stream_pending_data.remove(&close.stream_id); + stream_next_send_offset.remove(&close.stream_id); + stream_sent_data.remove(&close.stream_id); + stream_next_recv_offset.remove(&close.stream_id); + stream_recv_pending.remove(&close.stream_id); stream_writers.remove(&close.stream_id); agent_writers.remove(&close.stream_id); } @@ -2590,6 +2633,8 @@ async fn run_terminal( Some(ForwardEvent::Open { stream_id, target_host, target_port, writer, socks5_reply }) => { stream_writers.insert(stream_id, writer); stream_send_credit.insert(stream_id, STREAM_INITIAL_WINDOW); + stream_next_send_offset.entry(stream_id).or_insert(0); + stream_next_recv_offset.entry(stream_id).or_insert(0); if socks5_reply { pending_socks_replies.insert(stream_id); } @@ -2615,6 +2660,8 @@ async fn run_terminal( &opened_streams, &mut stream_send_credit, &mut stream_pending_data, + &mut stream_next_send_offset, + &mut stream_sent_data, ).await?; } Some(ForwardEvent::Close { stream_id }) => { @@ -2622,6 +2669,10 @@ async fn run_terminal( opened_streams.remove(&stream_id); stream_send_credit.remove(&stream_id); stream_pending_data.remove(&stream_id); + stream_next_send_offset.remove(&stream_id); + stream_sent_data.remove(&stream_id); + stream_next_recv_offset.remove(&stream_id); + stream_recv_pending.remove(&stream_id); stream_writers.remove(&stream_id); agent_writers.remove(&stream_id); send_stream_close(&socket, addr, &cred, &mut send_seq, stream_id).await?; @@ -2676,6 +2727,15 @@ async fn run_terminal( disconnect_status.apply(action)?; } } + _ = stream_retransmit_tick.tick() => { + retransmit_stream_data( + &socket, + addr, + &cred, + &mut send_seq, + &mut stream_sent_data, + ).await?; + } } } // Leave the terminal clean on exit: erase any lingering status line. @@ -2986,9 +3046,14 @@ async fn send_stream_data( cred: &CachedCredential, send_seq: &mut u64, stream_id: u64, + offset: u64, bytes: Vec, ) -> Result<()> { - let body = protocol::to_body(&StreamData { stream_id, bytes })?; + let body = protocol::to_body(&StreamData { + stream_id, + offset, + bytes, + })?; send_stream_packet(socket, addr, cred, send_seq, PacketKind::StreamData, &body).await } @@ -3002,6 +3067,8 @@ async fn queue_or_send_stream_data( opened_streams: &HashSet, stream_send_credit: &mut HashMap, stream_pending_data: &mut HashMap>>, + stream_next_send_offset: &mut HashMap, + stream_sent_data: &mut HashMap>, ) -> Result<()> { if !opened_streams.contains(&stream_id) || stream_send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len() @@ -3016,7 +3083,18 @@ async fn queue_or_send_stream_data( return Ok(()); } *stream_send_credit.entry(stream_id).or_default() -= bytes.len(); - send_stream_data(socket, addr, cred, send_seq, stream_id, bytes).await + let offset = *stream_next_send_offset.entry(stream_id).or_default(); + stream_next_send_offset.insert(stream_id, offset.saturating_add(bytes.len() as u64)); + stream_sent_data.entry(stream_id).or_default().insert( + offset, + PendingStreamChunk { + offset, + bytes: bytes.clone(), + last_sent: Instant::now(), + attempts: 1, + }, + ); + send_stream_data(socket, addr, cred, send_seq, stream_id, offset, bytes).await } async fn flush_stream_pending_data( @@ -3027,6 +3105,8 @@ async fn flush_stream_pending_data( stream_id: u64, stream_send_credit: &mut HashMap, stream_pending_data: &mut HashMap>>, + stream_next_send_offset: &mut HashMap, + stream_sent_data: &mut HashMap>, ) -> Result<()> { let Some(pending) = stream_pending_data.get_mut(&stream_id) else { return Ok(()); @@ -3041,7 +3121,18 @@ async fn flush_stream_pending_data( } let bytes = pending.pop_front().expect("pending front exists"); *stream_send_credit.entry(stream_id).or_default() -= bytes.len(); - send_stream_data(socket, addr, cred, send_seq, stream_id, bytes).await?; + let offset = *stream_next_send_offset.entry(stream_id).or_default(); + stream_next_send_offset.insert(stream_id, offset.saturating_add(bytes.len() as u64)); + stream_sent_data.entry(stream_id).or_default().insert( + offset, + PendingStreamChunk { + offset, + bytes: bytes.clone(), + last_sent: Instant::now(), + attempts: 1, + }, + ); + send_stream_data(socket, addr, cred, send_seq, stream_id, offset, bytes).await?; } if pending.is_empty() { stream_pending_data.remove(&stream_id); @@ -3049,9 +3140,103 @@ async fn flush_stream_pending_data( Ok(()) } -fn add_stream_credit(stream_send_credit: &mut HashMap, stream_id: u64, bytes: u32) { +async fn retransmit_stream_data( + socket: &UdpSocket, + addr: SocketAddr, + cred: &CachedCredential, + send_seq: &mut u64, + stream_sent_data: &mut HashMap>, +) -> Result<()> { + let now = Instant::now(); + let mut retransmit = Vec::new(); + for (stream_id, chunks) in stream_sent_data.iter_mut() { + for chunk in chunks.values_mut() { + if now.duration_since(chunk.last_sent) < Duration::from_millis(200) { + continue; + } + chunk.last_sent = now; + chunk.attempts = chunk.attempts.saturating_add(1); + retransmit.push((*stream_id, chunk.offset, chunk.bytes.clone())); + } + } + for (stream_id, offset, bytes) in retransmit { + send_stream_data(socket, addr, cred, send_seq, stream_id, offset, bytes).await?; + } + Ok(()) +} + +fn accept_stream_data( + stream_next_recv_offset: &mut HashMap, + stream_recv_pending: &mut HashMap>>, + data: StreamData, +) -> (Vec>, usize, u64) { + let stream_id = data.stream_id; + let expected = stream_next_recv_offset.entry(stream_id).or_insert(0); + if data.offset < *expected { + return (Vec::new(), 0, *expected); + } + if data.offset > *expected { + stream_recv_pending + .entry(stream_id) + .or_default() + .entry(data.offset) + .or_insert(data.bytes); + return (Vec::new(), 0, *expected); + } + + let mut writes = vec![data.bytes]; + let mut consumed = writes[0].len(); + *expected = expected.saturating_add(consumed as u64); + + while let Some(bytes) = stream_recv_pending + .entry(stream_id) + .or_default() + .remove(expected) + { + consumed = consumed.saturating_add(bytes.len()); + *expected = expected.saturating_add(bytes.len() as u64); + writes.push(bytes); + } + if stream_recv_pending + .get(&stream_id) + .is_some_and(BTreeMap::is_empty) + { + stream_recv_pending.remove(&stream_id); + } + (writes, consumed, *expected) +} + +fn ack_stream_data( + stream_sent_data: &mut HashMap>, + stream_send_credit: &mut HashMap, + stream_id: u64, + received_offset: u64, +) { + let Some(sent) = stream_sent_data.get_mut(&stream_id) else { + return; + }; + let acked_offsets: Vec = sent + .iter() + .filter_map(|(offset, chunk)| { + let end = chunk.offset.saturating_add(chunk.bytes.len() as u64); + (end <= received_offset).then_some(*offset) + }) + .collect(); + let mut acked_bytes = 0usize; + for offset in acked_offsets { + if let Some(chunk) = sent.remove(&offset) { + acked_bytes = acked_bytes.saturating_add(chunk.bytes.len()); + } + } + if sent.is_empty() { + stream_sent_data.remove(&stream_id); + } + add_stream_credit(stream_send_credit, stream_id, acked_bytes); +} + +fn add_stream_credit(stream_send_credit: &mut HashMap, stream_id: u64, bytes: usize) { let credit = stream_send_credit.entry(stream_id).or_default(); - *credit = credit.saturating_add(bytes as usize); + *credit = credit.saturating_add(bytes).min(STREAM_INITIAL_WINDOW); } async fn send_stream_window_adjust( @@ -3061,9 +3246,11 @@ async fn send_stream_window_adjust( send_seq: &mut u64, stream_id: u64, bytes: usize, + received_offset: u64, ) -> Result<()> { let body = protocol::to_body(&StreamWindowAdjust { stream_id, + received_offset, bytes: bytes.min(u32::MAX as usize) as u32, })?; send_stream_packet( diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index fa9bae1..d4c3eb6 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -20,7 +20,7 @@ use dosh::protocol::{ TicketAttachBody, TicketAttachEnvelope, TicketAttachOkEnvelope, }; use dosh::pty::{PtyHandle, PtyOutput, adopt_pty_from_fd, spawn_pty_session}; -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::net::{IpAddr, SocketAddr}; use std::os::unix::fs::PermissionsExt; use std::os::unix::net::UnixStream as StdUnixStream; @@ -375,6 +375,10 @@ struct ClientState { opened_streams: HashSet, stream_send_credit: HashMap, stream_pending_data: HashMap>>, + stream_next_send_offset: HashMap, + stream_sent_data: HashMap>, + stream_next_recv_offset: HashMap, + stream_recv_pending: HashMap>>, /// Current transport key epoch (0 = original handshake key). Bumped on rekey. epoch: u64, /// When the current epoch began, for the wall-clock rekey trigger. @@ -394,6 +398,14 @@ struct PendingFrame { attempts: u8, } +#[derive(Clone)] +struct PendingStreamChunk { + offset: u64, + bytes: Vec, + last_sent: Instant, + attempts: u32, +} + struct PendingNativeAuth { client: dosh::native::NativeClientHello, server: NativeServerHello, @@ -1115,6 +1127,10 @@ async fn handle_native_user_auth( opened_streams: HashSet::new(), stream_send_credit: HashMap::new(), stream_pending_data: HashMap::new(), + stream_next_send_offset: HashMap::new(), + stream_sent_data: HashMap::new(), + stream_next_recv_offset: HashMap::new(), + stream_recv_pending: HashMap::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, @@ -1260,6 +1276,10 @@ async fn handle_bootstrap_attach( opened_streams: HashSet::new(), stream_send_credit: HashMap::new(), stream_pending_data: HashMap::new(), + stream_next_send_offset: HashMap::new(), + stream_sent_data: HashMap::new(), + stream_next_recv_offset: HashMap::new(), + stream_recv_pending: HashMap::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, @@ -1379,6 +1399,10 @@ async fn handle_ticket_attach( opened_streams: HashSet::new(), stream_send_credit: HashMap::new(), stream_pending_data: HashMap::new(), + stream_next_send_offset: HashMap::new(), + stream_sent_data: HashMap::new(), + stream_next_recv_offset: HashMap::new(), + stream_recv_pending: HashMap::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, @@ -1745,6 +1769,14 @@ async fn handle_stream_open( client .stream_send_credit .insert(open.stream_id, STREAM_INITIAL_WINDOW); + client + .stream_next_send_offset + .entry(open.stream_id) + .or_insert(0); + client + .stream_next_recv_offset + .entry(open.stream_id) + .or_insert(0); } tokio::spawn(async move { @@ -1798,7 +1830,8 @@ async fn handle_stream_data( let (key, session_name) = find_client_decrypt_key(state, &packet.header)?; let body = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?; let data: StreamData = protocol::from_body(&body)?; - let writer = { + let stream_id = data.stream_id; + let (writer, writes, consumed, received_offset) = { let mut locked = state.lock().expect("server state poisoned"); let session = locked .sessions @@ -1813,17 +1846,27 @@ async fn handle_stream_data( } client.endpoint = peer; client.last_seen = Instant::now(); - client.stream_writers.get(&data.stream_id).cloned() + let writer = client.stream_writers.get(&data.stream_id).cloned(); + let (writes, consumed, received_offset) = accept_stream_data(client, data); + (writer, writes, consumed, received_offset) }; - let len = data.bytes.len(); if let Some(writer) = writer { - let _ = writer.send(data.bytes).await; + for bytes in writes { + let _ = writer.send(bytes).await; + } } // Always return flow-control credit, even when the stream's writer is already // gone (closed/unknown). The peer debited its send window for these bytes, so // skipping the adjust would wedge the stream at a permanent credit deficit. - send_stream_window_adjust_to_client(state, socket, packet.header.conn_id, data.stream_id, len) - .await?; + send_stream_window_adjust_to_client( + state, + socket, + packet.header.conn_id, + stream_id, + consumed, + received_offset, + ) + .await?; Ok(()) } @@ -1913,11 +1956,7 @@ async fn handle_stream_window_adjust( } client.endpoint = peer; client.last_seen = Instant::now(); - add_stream_credit( - &mut client.stream_send_credit, - adjust.stream_id, - adjust.bytes, - ); + ack_stream_data(client, adjust.stream_id, adjust.received_offset); } flush_stream_pending_data_to_client(state, socket, packet.header.conn_id, adjust.stream_id) .await @@ -1948,6 +1987,10 @@ async fn handle_stream_close( client.opened_streams.remove(&close.stream_id); client.stream_send_credit.remove(&close.stream_id); client.stream_pending_data.remove(&close.stream_id); + client.stream_next_send_offset.remove(&close.stream_id); + client.stream_sent_data.remove(&close.stream_id); + client.stream_next_recv_offset.remove(&close.stream_id); + client.stream_recv_pending.remove(&close.stream_id); Ok(()) } @@ -2312,8 +2355,16 @@ async fn queue_or_send_stream_data_to_client( return Ok(()); } *client.stream_send_credit.entry(stream_id).or_default() -= bytes.len(); + let offset = *client.stream_next_send_offset.entry(stream_id).or_default(); + client + .stream_next_send_offset + .insert(stream_id, offset.saturating_add(bytes.len() as u64)); client.send_seq += 1; - let body = protocol::to_body(&StreamData { stream_id, bytes })?; + let body = protocol::to_body(&StreamData { + stream_id, + offset, + bytes: bytes.clone(), + })?; let packet = protocol::encode_encrypted( PacketKind::StreamData, client_id, @@ -2323,6 +2374,19 @@ async fn queue_or_send_stream_data_to_client( SERVER_TO_CLIENT, &body, )?; + client + .stream_sent_data + .entry(stream_id) + .or_default() + .insert( + offset, + PendingStreamChunk { + offset, + bytes, + last_sent: Instant::now(), + attempts: 1, + }, + ); send = Some((client.endpoint, packet)); } send @@ -2364,8 +2428,16 @@ async fn flush_stream_pending_data_to_client( client.stream_pending_data.remove(&stream_id); } *client.stream_send_credit.entry(stream_id).or_default() -= bytes.len(); + let offset = *client.stream_next_send_offset.entry(stream_id).or_default(); + client + .stream_next_send_offset + .insert(stream_id, offset.saturating_add(bytes.len() as u64)); client.send_seq += 1; - let body = protocol::to_body(&StreamData { stream_id, bytes })?; + let body = protocol::to_body(&StreamData { + stream_id, + offset, + bytes: bytes.clone(), + })?; let packet = protocol::encode_encrypted( PacketKind::StreamData, client_id, @@ -2375,6 +2447,19 @@ async fn flush_stream_pending_data_to_client( SERVER_TO_CLIENT, &body, )?; + client + .stream_sent_data + .entry(stream_id) + .or_default() + .insert( + offset, + PendingStreamChunk { + offset, + bytes, + last_sent: Instant::now(), + attempts: 1, + }, + ); send = Some((client.endpoint, packet)); } send @@ -2386,9 +2471,72 @@ async fn flush_stream_pending_data_to_client( } } -fn add_stream_credit(stream_send_credit: &mut HashMap, stream_id: u64, bytes: u32) { +fn accept_stream_data(client: &mut ClientState, data: StreamData) -> (Vec>, usize, u64) { + let stream_id = data.stream_id; + let expected = client.stream_next_recv_offset.entry(stream_id).or_insert(0); + if data.offset < *expected { + return (Vec::new(), 0, *expected); + } + if data.offset > *expected { + client + .stream_recv_pending + .entry(stream_id) + .or_default() + .entry(data.offset) + .or_insert(data.bytes); + return (Vec::new(), 0, *expected); + } + + let mut writes = vec![data.bytes]; + let mut consumed = writes[0].len(); + *expected = expected.saturating_add(consumed as u64); + + while let Some(bytes) = client + .stream_recv_pending + .entry(stream_id) + .or_default() + .remove(expected) + { + consumed += bytes.len(); + *expected = expected.saturating_add(bytes.len() as u64); + writes.push(bytes); + } + if client + .stream_recv_pending + .get(&stream_id) + .is_some_and(BTreeMap::is_empty) + { + client.stream_recv_pending.remove(&stream_id); + } + (writes, consumed, *expected) +} + +fn ack_stream_data(client: &mut ClientState, stream_id: u64, received_offset: u64) { + let Some(sent) = client.stream_sent_data.get_mut(&stream_id) else { + return; + }; + let acked_offsets: Vec = sent + .iter() + .filter_map(|(offset, chunk)| { + let end = chunk.offset.saturating_add(chunk.bytes.len() as u64); + (end <= received_offset).then_some(*offset) + }) + .collect(); + let mut acked_bytes = 0usize; + for offset in acked_offsets { + if let Some(chunk) = sent.remove(&offset) { + acked_bytes = acked_bytes.saturating_add(chunk.bytes.len()); + } + } + if sent.is_empty() { + client.stream_sent_data.remove(&stream_id); + } + add_stream_credit(&mut client.stream_send_credit, stream_id, acked_bytes); +} + +fn add_stream_credit(stream_send_credit: &mut HashMap, stream_id: u64, bytes: usize) { let credit = stream_send_credit.entry(stream_id).or_default(); - *credit = credit.saturating_add(bytes as usize); + *credit = credit.saturating_add(bytes).min(STREAM_INITIAL_WINDOW); } async fn send_stream_window_adjust_to_client( @@ -2397,9 +2545,11 @@ async fn send_stream_window_adjust_to_client( client_id: [u8; 16], stream_id: u64, bytes: usize, + received_offset: u64, ) -> Result<()> { let body = protocol::to_body(&StreamWindowAdjust { stream_id, + received_offset, bytes: bytes.min(u32::MAX as usize) as u32, })?; send_stream_packet_to_client( @@ -2425,6 +2575,10 @@ async fn send_stream_close_to_client( client.opened_streams.remove(&stream_id); client.stream_send_credit.remove(&stream_id); client.stream_pending_data.remove(&stream_id); + client.stream_next_send_offset.remove(&stream_id); + client.stream_sent_data.remove(&stream_id); + client.stream_next_recv_offset.remove(&stream_id); + client.stream_recv_pending.remove(&stream_id); } } let body = protocol::to_body(&StreamClose { stream_id })?; @@ -2642,7 +2796,7 @@ async fn retransmit_pending( let now = Instant::now(); let mut sends = Vec::new(); for session in locked.sessions.values_mut() { - for client in session.clients.values_mut() { + for (client_id, client) in session.clients.iter_mut() { for pending in client.pending.iter_mut() { if pending.output_seq <= client.last_acked { continue; @@ -2655,6 +2809,39 @@ async fn retransmit_pending( sends.push((client.endpoint, pending.packet.clone())); } } + let stream_ids: Vec = client.stream_sent_data.keys().copied().collect(); + let mut retransmit_chunks = Vec::new(); + for stream_id in stream_ids { + let Some(chunks) = client.stream_sent_data.get_mut(&stream_id) else { + continue; + }; + for chunk in chunks.values_mut() { + if now.duration_since(chunk.last_sent) < Duration::from_millis(200) { + continue; + } + chunk.last_sent = now; + chunk.attempts = chunk.attempts.saturating_add(1); + retransmit_chunks.push((stream_id, chunk.offset, chunk.bytes.clone())); + } + } + for (stream_id, offset, bytes) in retransmit_chunks { + client.send_seq += 1; + let body = protocol::to_body(&StreamData { + stream_id, + offset, + bytes, + })?; + let packet = protocol::encode_encrypted( + PacketKind::StreamData, + *client_id, + client.send_seq, + client.last_acked, + &client.session_key, + SERVER_TO_CLIENT, + &body, + )?; + sends.push((client.endpoint, packet)); + } } } sends @@ -3015,6 +3202,10 @@ mod tests { opened_streams: HashSet::new(), stream_send_credit: HashMap::new(), stream_pending_data: HashMap::new(), + stream_next_send_offset: HashMap::new(), + stream_sent_data: HashMap::new(), + stream_next_recv_offset: HashMap::new(), + stream_recv_pending: HashMap::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, @@ -3211,6 +3402,10 @@ mod tests { opened_streams: HashSet::new(), stream_send_credit: HashMap::new(), stream_pending_data: HashMap::new(), + stream_next_send_offset: HashMap::new(), + stream_sent_data: HashMap::new(), + stream_next_recv_offset: HashMap::new(), + stream_recv_pending: HashMap::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, @@ -3300,6 +3495,10 @@ mod tests { opened_streams: HashSet::from([42]), stream_send_credit: HashMap::from([(42, 0)]), stream_pending_data: HashMap::new(), + stream_next_send_offset: HashMap::new(), + stream_sent_data: HashMap::new(), + stream_next_recv_offset: HashMap::new(), + stream_recv_pending: HashMap::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, diff --git a/src/protocol.rs b/src/protocol.rs index 1c0bae1..8e4ed82 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -38,12 +38,14 @@ pub fn is_implicit_session_name(name: &str) -> bool { } pub const MAGIC: &[u8; 4] = b"DOSH"; +// v4: added reliable stream offsets/acks to `StreamData` and +// `StreamWindowAdjust`. // v3: added `ForwardingKind::Agent` (SSH-agent forwarding). The new variant rides // inside `NativeUserAuth.requested_forwardings`, so a pre-agent peer would fail to // deserialize it; bumping the wire version makes such a peer answer with a clear // version-mismatch reject instead. Existing variants' bincode discriminants are // unchanged, so the bump is purely a compatibility gate. -pub const VERSION: u8 = 3; +pub const VERSION: u8 = 4; pub const HEADER_LEN: usize = 58; /// Stable, user-facing reason string the server puts in an `AttachReject` when a @@ -416,12 +418,14 @@ pub struct StreamOpenReject { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StreamData { pub stream_id: u64, + pub offset: u64, pub bytes: Vec, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StreamWindowAdjust { pub stream_id: u64, + pub received_offset: u64, pub bytes: u32, } diff --git a/tests/hostile_network.rs b/tests/hostile_network.rs index b6906db..020dcb7 100644 --- a/tests/hostile_network.rs +++ b/tests/hostile_network.rs @@ -20,7 +20,7 @@ //! tests are reproducible and fast. use std::fs; -use std::io::Read; +use std::io::{Read, Write}; use std::net::{SocketAddr, TcpListener, UdpSocket}; use std::process::{Child, Command, Stdio}; use std::sync::Arc; @@ -38,7 +38,7 @@ use dosh::native::{self, ForwardingKind, ForwardingRequest, NativeAuthOk, Native use dosh::protocol::{ self, AttachOk, AttachReject, CLIENT_TO_SERVER, Frame, Header, Input, NativeAuthOkBody, NativeClientHelloBody, NativeServerHelloBody, NativeUserAuthBody, PacketKind, ResumeRequest, - SERVER_TO_CLIENT, StreamData, StreamOpen, StreamOpenOk, + SERVER_TO_CLIENT, StreamData, StreamOpen, StreamOpenOk, StreamWindowAdjust, }; use ed25519_dalek::{SigningKey, VerifyingKey}; use rand::rngs::StdRng; @@ -627,6 +627,25 @@ fn wait_for_stream_open_ok( None } +fn wait_for_stream_data( + socket: &UdpSocket, + key: &[u8; 32], + stream_id: u64, + millis: u64, +) -> Option<(Header, StreamData)> { + let deadline = Instant::now() + Duration::from_millis(millis); + while Instant::now() < deadline { + if let Some((header, data)) = + recv_stream_packet::(socket, key, PacketKind::StreamData) + { + if data.stream_id == stream_id { + return Some((header, data)); + } + } + } + None +} + fn start_tcp_collector() -> (u16, Receiver>) { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let port = listener.local_addr().unwrap().port(); @@ -656,6 +675,34 @@ fn start_tcp_collector() -> (u16, Receiver>) { (port, rx) } +fn start_tcp_echo() -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + thread::spawn(move || { + let Ok((mut stream, _)) = listener.accept() else { + return; + }; + stream + .set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + let mut buf = [0u8; 1024]; + while Instant::now() < deadline { + match stream.read(&mut buf) { + Ok(0) => return, + Ok(n) => { + let _ = stream.write_all(&buf[..n]); + } + Err(ref err) + if err.kind() == std::io::ErrorKind::WouldBlock + || err.kind() == std::io::ErrorKind::TimedOut => {} + Err(_) => return, + } + } + }); + port +} + fn collect_tcp(rx: &Receiver>, millis: u64) -> Vec { let deadline = Instant::now() + Duration::from_millis(millis); let mut out = Vec::new(); @@ -895,6 +942,7 @@ fn forwarded_stream_data_survives_reorder_and_rejects_replay() { open_ok.seq, protocol::to_body(&StreamData { stream_id, + offset: 0, bytes: b"A".to_vec(), }) .unwrap(), @@ -909,6 +957,7 @@ fn forwarded_stream_data_survives_reorder_and_rejects_replay() { open_ok.seq, protocol::to_body(&StreamData { stream_id, + offset: 1, bytes: b"B".to_vec(), }) .unwrap(), @@ -916,8 +965,8 @@ fn forwarded_stream_data_survives_reorder_and_rejects_replay() { let mut seen = collect_tcp(&tcp_rx, 1500); assert!( - seen.contains(&b'A') && seen.contains(&b'B'), - "reordered stream bytes did not both reach TCP target: {:?}", + seen.starts_with(b"AB"), + "reordered stream bytes were not delivered once in-order to TCP target: {:?}", String::from_utf8_lossy(&seen) ); @@ -933,6 +982,7 @@ fn forwarded_stream_data_survives_reorder_and_rejects_replay() { CLIENT_TO_SERVER, &protocol::to_body(&StreamData { stream_id, + offset: 2, bytes: b"X".to_vec(), }) .unwrap(), @@ -959,6 +1009,94 @@ fn forwarded_stream_data_survives_reorder_and_rejects_replay() { ); } +#[test] +fn forwarded_stream_data_recovers_after_server_to_client_loss() { + let dir = tempfile::tempdir().unwrap(); + let port = free_udp_port(); + let config = write_server_config(&dir, port); + let signing_key = SigningKey::from_bytes(&[43u8; 32]); + authorize_ed25519_key(&dir, &signing_key); + let target_port = start_tcp_echo(); + let mut server = start_server(&dir, &config); + let relay = Relay::spawn(port, 0xA11CEu64); + + let attached = native_attach_through_relay( + &relay, + &signing_key, + vec![ForwardingRequest { + kind: ForwardingKind::Local, + bind_host: Some("127.0.0.1".to_string()), + listen_port: 0, + target_host: Some("127.0.0.1".to_string()), + target_port: Some(target_port), + }], + ); + let stream_id = 55; + let open = StreamOpen { + stream_id, + target_host: "127.0.0.1".to_string(), + target_port, + }; + send_stream_packet( + &attached.socket, + &relay, + &attached.ok, + PacketKind::StreamOpen, + 3, + attached.ok.initial_seq, + protocol::to_body(&open).unwrap(), + ); + let open_ok = + wait_for_stream_open_ok(&attached.socket, &attached.ok.session_key, stream_id, 3000) + .expect("stream did not open"); + + // Drop the first server->client echo and at least one retransmit tick. Once + // the link clears, the unacked stream chunk must be re-encrypted with a new + // packet sequence and delivered at the same stream offset. + relay.set_drop_s2c(100); + send_stream_packet( + &attached.socket, + &relay, + &attached.ok, + PacketKind::StreamData, + 4, + open_ok.seq, + protocol::to_body(&StreamData { + stream_id, + offset: 0, + bytes: b"PING".to_vec(), + }) + .unwrap(), + ); + thread::sleep(Duration::from_millis(300)); + relay.set_drop_s2c(0); + + let (_header, data) = + wait_for_stream_data(&attached.socket, &attached.ok.session_key, stream_id, 3000) + .expect("lost server->client stream data was not retransmitted"); + assert_eq!(data.offset, 0); + assert_eq!(data.bytes, b"PING"); + + send_stream_packet( + &attached.socket, + &relay, + &attached.ok, + PacketKind::StreamWindowAdjust, + 5, + open_ok.seq, + protocol::to_body(&StreamWindowAdjust { + stream_id, + received_offset: 4, + bytes: 4, + }) + .unwrap(), + ); + + drop(relay); + let _ = server.kill(); + let _ = server.wait(); +} + #[test] fn stale_packets_after_resume_are_ignored_not_fatal() { let dir = tempfile::tempdir().unwrap();