Merge pull request #472 from chhsiao90/handle-frames-on-closed-stream

ProtocolError or StreamClosedError depends on how the stream is closed by

Author: Lukasa

HISTORY.rst

@@ -42,6 +42,10 @@ Bugfixes
 - Resolved issue where the ``HTTP2-Settings`` header value sent by a client for
   plaintext upgrade would be ignored by ``initiate_upgrade_connection``, rather
   than have those settings applied appropriately.
+- Resolved an issue whereby certain frames received from a peer in the CLOSED
+  state would trigger connection errors when RFC 7540 says they should have
+  triggered stream errors instead. Added more detailed stream closure tracking
+  to ensure we don't throw away connections unnecessarily.
 
 
 2.5.2 (2017-01-27)

h2/connection.py

@@ -33,7 +33,7 @@
 )
 from .frame_buffer import FrameBuffer
 from .settings import Settings, SettingCodes
-from .stream import H2Stream
+from .stream import H2Stream, StreamClosedBy
 from .utilities import guard_increment_window
 from .windows import WindowManager
 
@@ -378,10 +378,13 @@ def __init__(self, client_side=True, header_encoding='utf-8', config=None):
         # Data that needs to be sent.
         self._data_to_send = b''
 
-        # Keeps track of streams that have been forcefully reset by this peer.
+        # Keeps track of how streams are closed.
         # Used to ensure that we don't blow up in the face of frames that were
-        # in flight when a stream was reset.
-        self._reset_streams = set()
+        # in flight when a RST_STREAM was sent.
+        # Also used to determine whether we should consider a frame received
+        # while a stream is closed as either a stream error or a connection
+        # error.
+        self._closed_streams = {}
 
         # The flow control window manager for the connection.
         self._inbound_flow_control_window_manager = WindowManager(
@@ -425,7 +428,8 @@ def _open_streams(self, remainder):
                 to_delete.append(stream_id)
 
         for stream_id in to_delete:
-            del self.streams[stream_id]
+            stream = self.streams.pop(stream_id)
+            self._closed_streams[stream_id] = stream.closed_by
 
         return count
 
@@ -1035,8 +1039,6 @@ def reset_stream(self, stream_id, error_code=0):
         stream = self._get_stream_by_id(stream_id)
         frames = stream.reset_stream(error_code)
         self._prepare_for_sending(frames)
-        self._reset_streams.add(stream_id)
-        del self.streams[stream_id]
 
     def close_connection(self, error_code=0, additional_data=None,
                          last_stream_id=None):
@@ -1488,27 +1490,37 @@ def _receive_frame(self, frame):
             # I don't love using __class__ here, maybe reconsider it.
             frames, events = self._frame_dispatch_table[frame.__class__](frame)
         except StreamClosedError as e:
-            # We need to send a RST_STREAM frame on behalf of the stream.
-            # The frame the stream wants to emit is already present in the
-            # exception.
-            # This does not require re-raising: it's an expected behaviour. The
-            # only time we don't do that is if this is a stream the user
-            # manually reset.
-            if frame.stream_id not in self._reset_streams:
+            # If the stream was closed by RST_STREAM, we just send a RST_STREAM
+            # to the remote peer. Otherwise, this is a connection error, and so
+            # we will re-raise to trigger one.
+            if self._stream_is_closed_by_reset(e.stream_id):
                 f = RstStreamFrame(e.stream_id)
                 f.error_code = e.error_code
                 self._prepare_for_sending([f])
                 events = e._events
             else:
-                events = []
+                raise
         except StreamIDTooLowError as e:
-            # The stream ID seems invalid. This is unlikely, so it's probably
-            # the case that this frame is actually for a stream that we've
-            # already reset and removed the state for. If it is, just swallow
-            # the error. If we didn't do that, re-raise.
-            if frame.stream_id not in self._reset_streams:
+            # The stream ID seems invalid. This may happen when the closed
+            # stream has been cleaned up, or when the remote peer has opened a
+            # new stream with a higher stream ID than this one, forcing it
+            # closed implicitly.
+            #
+            # Check how the stream was closed: depending on the mechanism, it
+            # is either a stream error or a connection error.
+            if self._stream_is_closed_by_reset(e.stream_id):
+                # Closed by RST_STREAM is a stream error.
+                f = RstStreamFrame(e.stream_id)
+                f.error_code = ErrorCodes.STREAM_CLOSED
+                self._prepare_for_sending([f])
+                events = []
+            elif self._stream_is_closed_by_end(e.stream_id):
+                # Closed by END_STREAM is a connection error.
+                raise StreamClosedError(e.stream_id)
+            else:
+                # Closed implicitly, also a connection error, but of type
+                # PROTOCOL_ERROR.
                 raise
-            events = []
         except KeyError as e:  # pragma: no cover
             # We don't have a function for handling this frame. Let's call this
             # a PROTOCOL_ERROR and exit.
@@ -1593,7 +1605,8 @@ def _receive_push_promise_frame(self, frame):
             # PROTOCOL_ERROR: pushing a stream on a naturally closed stream is
             # a real problem because it creates a brand new stream that the
             # remote peer now believes exists.
-            if frame.stream_id in self._reset_streams:
+            if (self._stream_closed_by(frame.stream_id) ==
+                    StreamClosedBy.SEND_RST_STREAM):
                 f = RstStreamFrame(frame.promised_stream_id)
                 f.error_code = ErrorCodes.REFUSED_STREAM
                 return [f], events
@@ -1608,11 +1621,19 @@ def _receive_push_promise_frame(self, frame):
         if (frame.stream_id % 2) == 0:
             raise ProtocolError("Cannot recursively push streams.")
 
-        frames, stream_events = stream.receive_push_promise_in_band(
-            frame.promised_stream_id,
-            pushed_headers,
-            self.config.header_encoding,
-        )
+        try:
+            frames, stream_events = stream.receive_push_promise_in_band(
+                frame.promised_stream_id,
+                pushed_headers,
+                self.config.header_encoding,
+            )
+        except StreamClosedError:
+            # The parent stream was reset by us, so we presume that
+            # PUSH_PROMISE was in flight when we reset the parent stream.
+            # So we just reset the new stream.
+            f = RstStreamFrame(frame.promised_stream_id)
+            f.error_code = ErrorCodes.REFUSED_STREAM
+            return [f], events
 
         new_stream = self._begin_new_stream(
             frame.promised_stream_id, AllowedStreamIDs.EVEN
@@ -1877,6 +1898,40 @@ def _stream_id_is_outbound(self, stream_id):
         """
         return (stream_id % 2 == int(self.config.client_side))
 
+    def _stream_closed_by(self, stream_id):
+        """
+        Returns how the stream was closed.
+
+        The return value will be either a member of
+        ``h2.stream.StreamClosedBy`` or ``None``. If ``None``, the stream was
+        closed implicitly by the peer opening a stream with a higher stream ID
+        before opening this one.
+        """
+        if stream_id in self.streams:
+            return self.streams[stream_id].closed_by
+        if stream_id in self._closed_streams:
+            return self._closed_streams[stream_id]
+        return None
+
+    def _stream_is_closed_by_reset(self, stream_id):
+        """
+        Returns ``True`` if the stream was closed by sending or receiving a
+        RST_STREAM frame. Returns ``False`` otherwise.
+        """
+        return self._stream_closed_by(stream_id) in (
+            StreamClosedBy.RECV_RST_STREAM, StreamClosedBy.SEND_RST_STREAM
+        )
+
+    def _stream_is_closed_by_end(self, stream_id):
+        """
+        Returns ``True`` if the stream was closed by sending or receiving an
+        END_STREAM flag in a HEADERS or DATA frame. Returns ``False``
+        otherwise.
+        """
+        return self._stream_closed_by(stream_id) in (
+            StreamClosedBy.RECV_END_STREAM, StreamClosedBy.SEND_END_STREAM
+        )
+
 
 def _add_frame_priority(frame, weight=None, depends_on=None, exclusive=None):
     """