Handle more H2 details

src/hpack/decoder.rs

@@ -36,6 +36,7 @@ pub enum DecoderError {
     IntegerOverflow,
     StringUnderflow,
     RepeatedPseudo,
+    UnexpectedEndOfStream,
 }
 
 enum Representation {
@@ -277,6 +278,10 @@ impl Decoder {
     fn decode_string(&mut self, buf: &mut Cursor<Bytes>) -> Result<Bytes, DecoderError> {
         const HUFF_FLAG: u8 = 0b10000000;
 
+        if !buf.has_remaining() {
+            return Err(DecoderError::UnexpectedEndOfStream);
+        }
+
         // The first bit in the first byte contains the huffman encoded flag.
         let huff = peek_u8(buf) & HUFF_FLAG == HUFF_FLAG;
 

src/proto/mod.rs

@@ -85,6 +85,9 @@ pub fn from_framed_write<T, P, B>(framed_write: FramedWrite<T, B::Buf>)
         .length_field_length(3)
         .length_adjustment(9)
         .num_skip(0) // Don't skip the header
+        // TODO: make this configurable and allow it to be changed during
+        // runtime.
+        .max_frame_length(frame::DEFAULT_MAX_FRAME_SIZE as usize)
         .new_read(framed_write);
 
     let codec = FramedRead::new(framed);

src/proto/streams/recv.rs

@@ -21,6 +21,9 @@ pub(super) struct Recv<B> {
     /// Connection level flow control governing received data
     flow_control: FlowControl,
 
+    /// The lowest stream ID that is still idle
+    next_stream_id: StreamId,
+
     /// Streams that have pending window updates
     /// TODO: don't use a VecDeque
     pending_window_updates: VecDeque<StreamId>,
@@ -50,12 +53,19 @@ struct Indices {
 }
 
 impl<B> Recv<B> where B: Buf {
-    pub fn new(config: &Config) -> Self {
+    pub fn new<P: Peer>(config: &Config) -> Self {
+        let next_stream_id = if P::is_server() {
+            1
+        } else {
+            2
+        };
+
         Recv {
             max_streams: config.max_remote_initiated,
             num_streams: 0,
             init_window_sz: config.init_remote_window_sz,
             flow_control: FlowControl::new(config.init_remote_window_sz),
+            next_stream_id: next_stream_id.into(),
             pending_window_updates: VecDeque::new(),
             pending_accept: store::List::new(),
             buffer: Buffer::new(),
@@ -129,6 +139,13 @@ impl<B> Recv<B> where B: Buf {
                 unimplemented!();
             }
 
+            if frame.stream_id() >= self.next_stream_id {
+                self.next_stream_id = frame.stream_id();
+                self.next_stream_id.increment();
+            } else {
+                return Err(ProtocolError.into());
+            }
+
             // Increment the number of concurrent streams
             self.inc_num_streams();
         }
@@ -246,6 +263,14 @@ impl<B> Recv<B> where B: Buf {
         Ok(())
     }
 
+    pub fn ensure_not_idle(&self, id: StreamId) -> Result<(), ConnectionError> {
+        if id >= self.next_stream_id {
+            return Err(ProtocolError.into());
+        }
+
+        Ok(())
+    }
+
     pub fn recv_reset(&mut self, frame: frame::Reset, stream: &mut Stream<B>)
         -> Result<(), ConnectionError>
     {

src/proto/streams/send.rs

@@ -273,6 +273,14 @@ impl<B> Send<B> where B: Buf {
         0
     }
 
+    pub fn ensure_not_idle(&self, id: StreamId) -> Result<(), ConnectionError> {
+        if id >= self.next_stream_id {
+            return Err(ProtocolError.into());
+        }
+
+        Ok(())
+    }
+
     pub fn dec_num_streams(&mut self) {
         self.num_streams -= 1;
     }

src/proto/streams/streams.rs

@@ -53,7 +53,7 @@ impl<B> Streams<B>
         Streams {
             inner: Arc::new(Mutex::new(Inner {
                 actions: Actions {
-                    recv: Recv::new(&config),
+                    recv: Recv::new::<P>(&config),
                     send: Send::new::<P>(&config),
                 },
                 store: Store::new(),
@@ -129,10 +129,17 @@ impl<B> Streams<B>
 
         let id = frame.stream_id();
 
+        if id.is_zero() {
+            return Err(ProtocolError.into());
+        }
+
         let mut stream = match me.store.find_mut(&id) {
             Some(stream) => stream,
-            // TODO: should this be an error?
-            None => return Ok(()),
+            None => {
+                // TODO: Are there other error cases?
+                me.actions.ensure_not_idle::<P>(id)?;
+                return Ok(());
+            }
         };
 
         me.actions.transition::<P, _, _>(stream, |actions, stream| {
@@ -165,6 +172,8 @@ impl<B> Streams<B>
             // considers closed. It's ok...
             if let Some(mut stream) = me.store.find_mut(&id) {
                 try!(me.actions.send.recv_stream_window_update(frame, &mut stream));
+            } else {
+                me.actions.recv.ensure_not_idle(id)?;
             }
         }
 
@@ -413,6 +422,16 @@ impl<B> Drop for Chunk<B>
 impl<B> Actions<B>
     where B: Buf,
 {
+    fn ensure_not_idle<P: Peer>(&mut self, id: StreamId)
+        -> Result<(), ConnectionError>
+    {
+        if self.is_local_init::<P>(id) {
+            self.send.ensure_not_idle(id)
+        } else {
+            self.recv.ensure_not_idle(id)
+        }
+    }
+
     fn dec_num_streams<P: Peer>(&mut self, id: StreamId) {
         if self.is_local_init::<P>(id) {
             self.send.dec_num_streams();