fix(transport): don't consider remote stream illegal (#2358)

#2269 made a "stream control or data frame for a stream that is ours to initiate but we haven't yet" illegal.

``` rust
       let existed = !stream_id.is_remote_initiated(self.role)
            && self.local_stream_limits[stream_id.stream_type()].used() > stream_id.index();
```

Problem is, when `is_remote_initiated` is `true`, then `existed` is `false`, which leads to a false positive for remote initiated streams.

Author: mxinden

neqo-transport/src/connection/tests/stream.rs

@@ -539,23 +539,23 @@ fn do_not_accept_data_after_stop_sending() {
     );
 }
 
+struct Writer(Vec<u64>);
+
+impl crate::connection::test_internal::FrameWriter for Writer {
+    fn write_frames(&mut self, builder: &mut PacketBuilder) {
+        builder.write_varint_frame(&self.0);
+    }
+}
+
 #[test]
 /// Server sends a number of stream-related frames for a client-initiated stream that is not yet
 /// created. This should cause the client to close the connection.
 fn illegal_stream_related_frames() {
-    struct IllegalWriter(Vec<u64>);
-
-    impl crate::connection::test_internal::FrameWriter for IllegalWriter {
-        fn write_frames(&mut self, builder: &mut PacketBuilder) {
-            builder.write_varint_frame(&self.0);
-        }
-    }
-
     fn test_with_illegal_frame(frame: &[u64]) {
         let mut client = default_client();
         let mut server = default_server();
         connect(&mut client, &mut server);
-        let dgram = send_with_extra(&mut server, IllegalWriter(frame.to_vec()), now());
+        let dgram = send_with_extra(&mut server, Writer(frame.to_vec()), now());
         client.process_input(dgram, now());
         assert!(client.state().closed());
     }
@@ -576,6 +576,47 @@ fn illegal_stream_related_frames() {
     }
 }
 
+#[test]
+/// Regression <https://github.com/mozilla/neqo/pull/2358>.
+fn legal_out_of_order_frame_on_remote_initiated_closed_stream() {
+    const REQUEST: &[u8] = b"ping";
+    let mut client = default_client();
+    let mut server = default_server();
+    connect(&mut client, &mut server);
+
+    // Client sends request and closes stream.
+    let stream_id = client.stream_create(StreamType::BiDi).unwrap();
+    _ = client.stream_send(stream_id, REQUEST).unwrap();
+    client.stream_close_send(stream_id).unwrap();
+    let dgram = client.process_output(now()).dgram();
+
+    // Server reads request and closes stream.
+    server.process_input(dgram.unwrap(), now());
+    let mut buf = [0; REQUEST.len()];
+    server.stream_recv(stream_id, &mut buf).unwrap();
+    server.stream_close_send(stream_id).unwrap();
+    let dgram = server.process_output(now()).dgram();
+    client.process_input(dgram.unwrap(), now());
+
+    // Client ACKs server's close stream, thus server forgetting about stream.
+    let dgram = send_something(&mut client, now());
+    let dgram = server.process(Some(dgram), now()).dgram();
+    client.process_input(dgram.unwrap(), now());
+
+    // Deliver an out-of-order `FRAME_TYPE_MAX_STREAM_DATA` on forgotten stream.
+    let dgram = send_with_extra(
+        &mut client,
+        Writer(vec![FRAME_TYPE_MAX_STREAM_DATA, stream_id.as_u64(), 0, 0]),
+        now(),
+    );
+    server.process_input(dgram, now());
+
+    assert!(
+        !server.state().closed(),
+        "expect server to ignore out-of-order frame on forgotten stream"
+    );
+}
+
 #[test]
 // Server sends stop_sending, the client simultaneous sends reset.
 fn simultaneous_stop_sending_and_reset() {

neqo-transport/src/streams.rs

@@ -411,13 +411,18 @@ impl Streams {
         stream_id: StreamId,
     ) -> Res<(Option<&mut SendStream>, Option<&mut RecvStream>)> {
         self.ensure_created_if_remote(stream_id)?;
-        // Has this local stream existed in the past, i.e., is its index lower than the number of
-        // used streams?
-        let existed = !stream_id.is_remote_initiated(self.role)
-            && self.local_stream_limits[stream_id.stream_type()].used() > stream_id.index();
         let ss = self.send.get_mut(stream_id).ok();
         let rs = self.recv.get_mut(stream_id).ok();
-        if ss.is_none() && rs.is_none() && !existed {
+        // If it is:
+        // - neither a known send nor receive stream,
+        // - and it must be locally initiated,
+        // - and its index is larger than the local used stream limit,
+        // then it is an illegal stream.
+        if ss.is_none()
+            && rs.is_none()
+            && !stream_id.is_remote_initiated(self.role)
+            && self.local_stream_limits[stream_id.stream_type()].used() <= stream_id.index()
+        {
             return Err(Error::StreamStateError);
         }
         Ok((ss, rs))