[stream] check final size does not change

src/aioquic/quic/connection.py

@@ -43,7 +43,7 @@
     QuicPacketBuilderStop,
 )
 from .recovery import K_GRANULARITY, QuicPacketRecovery, QuicPacketSpace
-from .stream import QuicStream
+from .stream import FinalSizeError, QuicStream
 
 logger = logging.getLogger("quic")
 
@@ -1824,10 +1824,16 @@ def _handle_reset_stream_frame(
             error_code,
             final_size,
         )
-        stream.handle_reset(final_size=final_size)
-        self._events.append(
-            events.StreamReset(error_code=error_code, stream_id=stream_id)
-        )
+        try:
+            event = stream.handle_reset(error_code=error_code, final_size=final_size)
+        except FinalSizeError as exc:
+            raise QuicConnectionError(
+                error_code=QuicErrorCode.FINAL_SIZE_ERROR,
+                frame_type=frame_type,
+                reason_phrase=str(exc),
+            )
+        if event is not None:
+            self._events.append(event)
         self._local_max_data.used += newly_received
 
     def _handle_retire_connection_id_frame(
@@ -1947,7 +1953,14 @@ def _handle_stream_frame(
             )
 
         # process data
-        event = stream.add_frame(frame)
+        try:
+            event = stream.add_frame(frame)
+        except FinalSizeError as exc:
+            raise QuicConnectionError(
+                error_code=QuicErrorCode.FINAL_SIZE_ERROR,
+                frame_type=frame_type,
+                reason_phrase=str(exc),
+            )
         if event is not None:
             self._events.append(event)
         self._local_max_data.used += newly_received

src/aioquic/quic/stream.py

@@ -1,11 +1,15 @@
 from typing import Optional
 
 from . import events
-from .packet import QuicResetStreamFrame, QuicStreamFrame
+from .packet import QuicErrorCode, QuicResetStreamFrame, QuicStreamFrame
 from .packet_builder import QuicDeliveryState
 from .rangeset import RangeSet
 
 
+class FinalSizeError(Exception):
+    pass
+
+
 class QuicStream:
     def __init__(
         self,
@@ -20,8 +24,8 @@ def __init__(
         self.send_buffer_is_empty = True
 
         self._recv_buffer = bytearray()
-        self._recv_buffer_fin: Optional[int] = None
         self._recv_buffer_start = 0  # the offset for the start of the buffer
+        self._recv_final_size: Optional[int] = None
         self._recv_highest = 0  # the highest offset ever seen
         self._recv_ranges = RangeSet()
 
@@ -57,10 +61,13 @@ def add_frame(self, frame: QuicStreamFrame) -> Optional[events.StreamDataReceive
         frame_end = frame.offset + count
 
         # we should receive no more data beyond FIN!
-        if self._recv_buffer_fin is not None and frame_end > self._recv_buffer_fin:
-            raise Exception("Data received beyond FIN")
+        if self._recv_final_size is not None:
+            if frame_end > self._recv_final_size:
+                raise FinalSizeError("Data received beyond final size")
+            elif frame.fin and frame_end != self._recv_final_size:
+                raise FinalSizeError("Cannot change final size")
         if frame.fin:
-            self._recv_buffer_fin = frame_end
+            self._recv_final_size = frame_end
         if frame_end > self._recv_highest:
             self._recv_highest = frame_end
 
@@ -89,7 +96,7 @@ def add_frame(self, frame: QuicStreamFrame) -> Optional[events.StreamDataReceive
 
         # return data from the front of the buffer
         data = self._pull_data()
-        end_stream = self._recv_buffer_start == self._recv_buffer_fin
+        end_stream = self._recv_buffer_start == self._recv_final_size
         if data or end_stream:
             return events.StreamDataReceived(
                 data=data, end_stream=end_stream, stream_id=self.__stream_id
@@ -129,11 +136,16 @@ def next_send_offset(self) -> int:
         except IndexError:
             return self._send_buffer_stop
 
-    def handle_reset(self, final_size: int) -> None:
+    def handle_reset(
+        self, *, final_size: int, error_code: int = QuicErrorCode.NO_ERROR
+    ) -> Optional[events.StreamReset]:
         """
         Handle an abrupt termination of the receiving part of the QUIC stream.
         """
-        self._recv_buffer_fin = final_size
+        if self._recv_final_size is not None and final_size != self._recv_final_size:
+            raise FinalSizeError("Cannot change final size")
+        self._recv_final_size = final_size
+        return events.StreamReset(error_code=error_code, stream_id=self.__stream_id)
 
     def get_frame(
         self, max_size: int, max_offset: Optional[int] = None

tests/test_connection.py

@@ -1431,6 +1431,44 @@ def test_handle_reset_stream_frame(self):
             self.assertEqual(event.error_code, QuicErrorCode.INTERNAL_ERROR)
             self.assertEqual(event.stream_id, stream_id)
 
+    def test_handle_reset_stream_frame_final_size_error(self):
+        stream_id = 0
+        with client_and_server() as (client, server):
+            # client creates bidirectional stream
+            client.send_stream_data(stream_id=stream_id, data=b"hello")
+            consume_events(client)
+
+            # client receives RESET_STREAM at offset 8
+            client._handle_reset_stream_frame(
+                client_receive_context(client),
+                QuicFrameType.RESET_STREAM,
+                Buffer(
+                    data=encode_uint_var(stream_id)
+                    + encode_uint_var(QuicErrorCode.NO_ERROR)
+                    + encode_uint_var(8)
+                ),
+            )
+
+            event = client.next_event()
+            self.assertEqual(type(event), events.StreamReset)
+            self.assertEqual(event.error_code, QuicErrorCode.NO_ERROR)
+            self.assertEqual(event.stream_id, stream_id)
+
+            # client receives RESET_STREAM at offset 5
+            with self.assertRaises(QuicConnectionError) as cm:
+                client._handle_reset_stream_frame(
+                    client_receive_context(client),
+                    QuicFrameType.RESET_STREAM,
+                    Buffer(
+                        data=encode_uint_var(stream_id)
+                        + encode_uint_var(QuicErrorCode.NO_ERROR)
+                        + encode_uint_var(5)
+                    ),
+                )
+            self.assertEqual(cm.exception.error_code, QuicErrorCode.FINAL_SIZE_ERROR)
+            self.assertEqual(cm.exception.frame_type, QuicFrameType.RESET_STREAM)
+            self.assertEqual(cm.exception.reason_phrase, "Cannot change final size")
+
     def test_handle_reset_stream_frame_over_max_data(self):
         stream_id = 0
         with client_and_server() as (client, server):
@@ -1586,6 +1624,37 @@ def test_handle_stop_sending_frame_receive_only(self):
             self.assertEqual(cm.exception.frame_type, QuicFrameType.STOP_SENDING)
             self.assertEqual(cm.exception.reason_phrase, "Stream is receive-only")
 
+    def test_handle_stream_frame_final_size_error(self):
+        with client_and_server() as (client, server):
+            frame_type = QuicFrameType.STREAM_BASE | 7
+            stream_id = 1
+
+            # client receives FIN at offset 8
+            client._handle_stream_frame(
+                client_receive_context(client),
+                frame_type,
+                Buffer(
+                    data=encode_uint_var(stream_id)
+                    + encode_uint_var(8)
+                    + encode_uint_var(0)
+                ),
+            )
+
+            # client receives FIN at offset 5
+            with self.assertRaises(QuicConnectionError) as cm:
+                client._handle_stream_frame(
+                    client_receive_context(client),
+                    frame_type,
+                    Buffer(
+                        data=encode_uint_var(stream_id)
+                        + encode_uint_var(5)
+                        + encode_uint_var(0)
+                    ),
+                )
+            self.assertEqual(cm.exception.error_code, QuicErrorCode.FINAL_SIZE_ERROR)
+            self.assertEqual(cm.exception.frame_type, frame_type)
+            self.assertEqual(cm.exception.reason_phrase, "Cannot change final size")
+
     def test_handle_stream_frame_over_largest_offset(self):
         with client_and_server() as (client, server):
             # client receives offset + length > 2^62 - 1

tests/test_stream.py

@@ -1,9 +1,9 @@
 from unittest import TestCase
 
-from aioquic.quic.events import StreamDataReceived
+from aioquic.quic.events import StreamDataReceived, StreamReset
 from aioquic.quic.packet import QuicErrorCode, QuicStreamFrame
 from aioquic.quic.packet_builder import QuicDeliveryState
-from aioquic.quic.stream import QuicStream
+from aioquic.quic.stream import FinalSizeError, QuicStream
 
 
 class QuicStreamTest(TestCase):
@@ -153,10 +153,17 @@ def test_recv_fin_out_of_order(self):
 
     def test_recv_fin_then_data(self):
         stream = QuicStream(stream_id=0)
-        stream.add_frame(QuicStreamFrame(offset=0, data=b"", fin=True))
-        with self.assertRaises(Exception) as cm:
+        stream.add_frame(QuicStreamFrame(offset=0, data=b"0123", fin=True))
+
+        # data beyond final size
+        with self.assertRaises(FinalSizeError) as cm:
             stream.add_frame(QuicStreamFrame(offset=0, data=b"01234567"))
-        self.assertEqual(str(cm.exception), "Data received beyond FIN")
+        self.assertEqual(str(cm.exception), "Data received beyond final size")
+
+        # final size would be lowered
+        with self.assertRaises(FinalSizeError) as cm:
+            stream.add_frame(QuicStreamFrame(offset=0, data=b"01", fin=True))
+        self.assertEqual(str(cm.exception), "Cannot change final size")
 
     def test_recv_fin_twice(self):
         stream = QuicStream(stream_id=0)
@@ -181,6 +188,43 @@ def test_recv_fin_without_data(self):
             StreamDataReceived(data=b"", end_stream=True, stream_id=0),
         )
 
+    def test_recv_reset(self):
+        stream = QuicStream(stream_id=0)
+        self.assertEqual(
+            stream.handle_reset(final_size=4),
+            StreamReset(error_code=QuicErrorCode.NO_ERROR, stream_id=0),
+        )
+
+    def test_recv_reset_after_fin(self):
+        stream = QuicStream(stream_id=0)
+        stream.add_frame(QuicStreamFrame(offset=0, data=b"0123", fin=True)),
+        self.assertEqual(
+            stream.handle_reset(final_size=4),
+            StreamReset(error_code=QuicErrorCode.NO_ERROR, stream_id=0),
+        )
+
+    def test_recv_reset_twice(self):
+        stream = QuicStream(stream_id=0)
+        self.assertEqual(
+            stream.handle_reset(final_size=4),
+            StreamReset(error_code=QuicErrorCode.NO_ERROR, stream_id=0),
+        )
+        self.assertEqual(
+            stream.handle_reset(final_size=4),
+            StreamReset(error_code=QuicErrorCode.NO_ERROR, stream_id=0),
+        )
+
+    def test_recv_reset_twice_final_size_error(self):
+        stream = QuicStream(stream_id=0)
+        self.assertEqual(
+            stream.handle_reset(final_size=4),
+            StreamReset(error_code=QuicErrorCode.NO_ERROR, stream_id=0),
+        )
+
+        with self.assertRaises(FinalSizeError) as cm:
+            stream.handle_reset(final_size=5)
+        self.assertEqual(str(cm.exception), "Cannot change final size")
+
     def test_send_data(self):
         stream = QuicStream()
         self.assertEqual(stream.next_send_offset, 0)