diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index de41bcf..a7cd71c 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -8110,6 +8110,14 @@ fn accept_stream_data( ) -> (Vec>, usize, u64) { let stream_id = data.stream_id; let expected = stream_next_recv_offset.entry(stream_id).or_insert(0); + if !dosh::transport::stream_data_within_receive_window( + *expected, + data.offset, + data.bytes.len(), + STREAM_INITIAL_WINDOW, + ) { + return (Vec::new(), 0, *expected); + } if data.offset < *expected { return (Vec::new(), 0, *expected); } diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index 649a861..1bd36e3 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -3509,6 +3509,14 @@ async fn flush_stream_pending_data_to_client( 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 !dosh::transport::stream_data_within_receive_window( + *expected, + data.offset, + data.bytes.len(), + STREAM_INITIAL_WINDOW, + ) { + return (Vec::new(), 0, *expected); + } if data.offset < *expected { return (Vec::new(), 0, *expected); } diff --git a/src/transport.rs b/src/transport.rs index fa401e7..5934315 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -630,6 +630,14 @@ impl StreamMux { fn accept_data(&mut self, data: StreamData) -> (Vec>, usize, u64) { let stream_id = data.stream_id; let expected = self.next_recv_offset.entry(stream_id).or_insert(0); + if !stream_data_within_receive_window( + *expected, + data.offset, + data.bytes.len(), + self.config.initial_window, + ) { + return (Vec::new(), 0, *expected); + } if data.offset < *expected { return (Vec::new(), 0, *expected); } @@ -769,6 +777,25 @@ pub fn observe_retransmit_rtt(srtt: &mut Option, sample: Duration) { }); } +pub fn stream_data_within_receive_window( + expected_offset: u64, + offset: u64, + len: usize, + receive_window: usize, +) -> bool { + let Ok(len) = u64::try_from(len) else { + return false; + }; + let window = receive_window as u64; + if len > window { + return false; + } + let Some(end) = offset.checked_add(len) else { + return false; + }; + end <= expected_offset.saturating_add(window) +} + pub struct DoshTransport { socket: Arc, role: SessionRole, @@ -1205,6 +1232,73 @@ mod tests { assert_eq!(second.received_offset, 10); } + #[test] + fn far_future_data_is_not_buffered_beyond_receive_window() { + let mut mux = StreamMux::new(TransportConfig { + initial_window: 5, + ..TransportConfig::default() + }); + mux.accept_open(StreamOpen { + stream_id: 3, + target_host: "@dosh-test".to_string(), + target_port: 0, + }) + .unwrap(); + + let too_far = mux + .handle_data(StreamData { + stream_id: 3, + offset: 5, + bytes: b"!".to_vec(), + }) + .unwrap() + .unwrap(); + assert!(too_far.chunks.is_empty()); + assert_eq!(too_far.consumed, 0); + assert_eq!(too_far.received_offset, 0); + assert!(!mux.recv_pending.contains_key(&3)); + + let in_window = mux + .handle_data(StreamData { + stream_id: 3, + offset: 4, + bytes: b"o".to_vec(), + }) + .unwrap() + .unwrap(); + assert!(in_window.chunks.is_empty()); + assert_eq!(in_window.received_offset, 0); + assert!( + mux.recv_pending + .get(&3) + .is_some_and(|pending| pending.contains_key(&4)) + ); + + let oversized_current = mux + .handle_data(StreamData { + stream_id: 3, + offset: 0, + bytes: b"abcdef".to_vec(), + }) + .unwrap() + .unwrap(); + assert!(oversized_current.chunks.is_empty()); + assert_eq!(oversized_current.received_offset, 0); + } + + #[test] + fn receive_window_helper_rejects_overflowing_ranges() { + assert!(stream_data_within_receive_window(0, 4, 1, 5)); + assert!(!stream_data_within_receive_window(0, 5, 1, 5)); + assert!(!stream_data_within_receive_window(0, 0, 6, 5)); + assert!(!stream_data_within_receive_window( + u64::MAX - 1, + u64::MAX, + 1, + 5 + )); + } + #[test] fn window_adjust_acks_sent_data_and_releases_credit() { let mut mux = StreamMux::new(TransportConfig {