From ccd30fd16d6eb2cf3d0d72182ebb84dadc4a61fb Mon Sep 17 00:00:00 2001 From: Dirkjan Ochtman Date: Wed, 28 Aug 2019 08:34:43 +0200 Subject: [PATCH] Don't panic from unknown StreamIds --- quinn-proto/src/connection.rs | 20 ++++++++++++++------ quinn-proto/src/lib.rs | 2 +- quinn-proto/src/streams.rs | 6 ++++++ quinn-proto/src/tests/mod.rs | 15 +++++++++++---- quinn/src/streams.rs | 21 ++++++++++++++++----- 5 files changed, 48 insertions(+), 16 deletions(-) diff --git a/quinn-proto/src/connection.rs b/quinn-proto/src/connection.rs index 220f116fe..133636e12 100644 --- a/quinn-proto/src/connection.rs +++ b/quinn-proto/src/connection.rs @@ -23,7 +23,7 @@ use crate::shared::{ EndpointEvent, EndpointEventInner, IssuedCid, ServerConfig, TransportConfig, }; use crate::spaces::{CryptoSpace, PacketSpace, Retransmits, SentPacket}; -use crate::streams::{self, FinishError, ReadError, Streams, WriteError}; +use crate::streams::{self, FinishError, ReadError, Streams, UnknownStream, WriteError}; use crate::timer::{Timer, TimerKind, TimerTable}; use crate::transport_parameters::{self, TransportParameters}; use crate::{ @@ -818,8 +818,11 @@ where ); } - fn queue_stream_data(&mut self, stream: StreamId, data: Bytes) { - let ss = self.streams.send_mut(stream).unwrap(); + fn queue_stream_data(&mut self, stream: StreamId, data: Bytes) -> Result<(), WriteError> { + let ss = self + .streams + .send_mut(stream) + .ok_or(WriteError::UnknownStream)?; assert_eq!(ss.state, streams::SendState::Ready); let offset = ss.offset; ss.offset += data.len() as u64; @@ -835,6 +838,7 @@ where data, id: stream, }); + Ok(()) } /// Abandon transmitting data on a stream @@ -2656,12 +2660,15 @@ where } /// Signal to the peer that it should stop sending on the given recv stream - pub fn stop_sending(&mut self, id: StreamId, error_code: VarInt) { + pub fn stop_sending(&mut self, id: StreamId, error_code: VarInt) -> Result<(), UnknownStream> { assert!( id.directionality() == Directionality::Bi || id.initiator() != self.side, "only streams supporting incoming data may be stopped" ); - let stream = self.streams.recv_mut(id).unwrap(); + let stream = self + .streams + .recv_mut(id) + .ok_or(UnknownStream { _private: () })?; // Only bother if there's data we haven't received yet if !stream.is_finished() { let space = &mut self.spaces[SpaceId::Data as usize]; @@ -2670,6 +2677,7 @@ where .stop_sending .push(frame::StopSending { id, error_code }); } + Ok(()) } fn congestion_blocked(&self) -> bool { @@ -2846,7 +2854,7 @@ where self.config.send_window - self.unacked_data, ); let n = conn_budget.min(stream_budget).min(data.len() as u64) as usize; - self.queue_stream_data(stream, (&data[0..n]).into()); + self.queue_stream_data(stream, (&data[0..n]).into())?; trace!( self.log, "wrote {len} bytes to {stream}", diff --git a/quinn-proto/src/lib.rs b/quinn-proto/src/lib.rs index f9de7ef40..3ce9923a4 100644 --- a/quinn-proto/src/lib.rs +++ b/quinn-proto/src/lib.rs @@ -58,7 +58,7 @@ pub use crate::shared::{ }; mod streams; -pub use crate::streams::{FinishError, ReadError, WriteError}; +pub use crate::streams::{FinishError, ReadError, UnknownStream, WriteError}; mod transport_error; pub use crate::transport_error::{Code as TransportErrorCode, Error as TransportError}; diff --git a/quinn-proto/src/streams.rs b/quinn-proto/src/streams.rs index 3baa3758c..9d66287a7 100644 --- a/quinn-proto/src/streams.rs +++ b/quinn-proto/src/streams.rs @@ -513,3 +513,9 @@ pub enum FinishError { #[error(display = "unknown stream")] UnknownStream, } + +/// Unknown stream ID +#[derive(Debug)] +pub struct UnknownStream { + pub(crate) _private: (), +} diff --git a/quinn-proto/src/tests/mod.rs b/quinn-proto/src/tests/mod.rs index 6a07bc2e2..a654de87b 100644 --- a/quinn-proto/src/tests/mod.rs +++ b/quinn-proto/src/tests/mod.rs @@ -272,7 +272,9 @@ fn stop_stream() { info!(pair.log, "stopping stream"); const ERROR: VarInt = VarInt(42); - pair.server_conn_mut(server_ch).stop_sending(s, ERROR); + pair.server_conn_mut(server_ch) + .stop_sending(s, ERROR) + .unwrap(); pair.drive(); assert_matches!( @@ -904,7 +906,8 @@ fn stop_opens_bidi() { .connections .get_mut(&server_conn) .unwrap() - .stop_sending(s, ERROR); + .stop_sending(s, ERROR) + .unwrap(); pair.drive(); assert_matches!( @@ -1090,7 +1093,9 @@ fn stop_before_finish() { info!(pair.log, "stopping stream"); const ERROR: VarInt = VarInt(42); - pair.server_conn_mut(server_ch).stop_sending(s, ERROR); + pair.server_conn_mut(server_ch) + .stop_sending(s, ERROR) + .unwrap(); pair.drive(); assert_matches!( @@ -1115,7 +1120,9 @@ fn stop_during_finish() { assert_matches!(pair.server_conn_mut(server_ch).accept(), Some(stream) if stream == s); info!(pair.log, "stopping and finishing stream"); const ERROR: VarInt = VarInt(42); - pair.server_conn_mut(server_ch).stop_sending(s, ERROR); + pair.server_conn_mut(server_ch) + .stop_sending(s, ERROR) + .unwrap(); pair.drive_server(); pair.client_conn_mut(client_ch).finish(s).unwrap(); pair.drive_client(); diff --git a/quinn/src/streams.rs b/quinn/src/streams.rs index a3649ea4e..ff2a7f5d9 100644 --- a/quinn/src/streams.rs +++ b/quinn/src/streams.rs @@ -6,7 +6,7 @@ use err_derive::Error; use futures::sync::oneshot; use futures::{task, try_ready}; use futures::{Async, Future, Poll}; -use proto::{ConnectionError, StreamId}; +use proto::{self, ConnectionError, StreamId}; use tokio_io::{AsyncRead, AsyncWrite}; use crate::connection::ConnectionRef; @@ -333,14 +333,15 @@ impl RecvStream { /// /// Has no effect if the incoming stream already finished, even if the local application hasn't /// yet read all buffered data. - pub fn stop(&mut self, error_code: VarInt) { + pub fn stop(&mut self, error_code: VarInt) -> Result<(), UnknownStream> { let mut conn = self.conn.lock().unwrap(); if self.is_0rtt && conn.check_0rtt().is_err() { - return; + return Ok(()); } - conn.inner.stop_sending(self.stream, error_code); + conn.inner.stop_sending(self.stream, error_code)?; conn.notify(); self.all_data_read = true; + Ok(()) } } @@ -422,7 +423,8 @@ impl Drop for RecvStream { return; } if !self.all_data_read { - conn.inner.stop_sending(self.stream, 0u32.into()); + // Ignore UnknownStream errors + let _ = conn.inner.stop_sending(self.stream, 0u32.into()); conn.notify(); } } @@ -499,3 +501,12 @@ impl From for io::Error { io::Error::new(kind, x) } } + +#[derive(Debug)] +pub struct UnknownStream {} + +impl From for UnknownStream { + fn from(_: proto::UnknownStream) -> Self { + UnknownStream {} + } +}