speculative HTTP/2 fixes, refs #4451 (#4464)

mitmproxy/proxy/layers/http/__init__.py

@@ -461,7 +461,7 @@ class HttpStream(layer.Layer):
             )
 
         if 200 <= self.flow.response.status_code < 300:
-            yield SendHttp(ResponseHeaders(self.stream_id, self.flow.response), self.context.client)
+            yield SendHttp(ResponseHeaders(self.stream_id, self.flow.response, True), self.context.client)
             yield SendHttp(ResponseEndOfMessage(self.stream_id), self.context.client)
             self.child_layer = self.child_layer or layer.NextLayer(self.context)
             yield from self.child_layer.handle_event(events.Start())

mitmproxy/proxy/layers/http/_http2.py

@@ -109,6 +109,7 @@ class Http2Connection(HttpConnection):
                     stream: h2.stream.H2Stream = self.h2_conn.streams[event.stream_id]
                     send_error_message = (
                         isinstance(event, ResponseProtocolError)
+                        and self.is_open_for_us(event.stream_id)
                         and not stream.state_machine.headers_sent
                         and event.code != status_codes.NO_RESPONSE
                     )
@@ -291,6 +292,7 @@ class Http2Server(Http2Connection):
                 self.h2_conn.send_headers(
                     event.stream_id,
                     headers,
+                    end_stream=event.end_stream,
                 )
                 yield SendData(self.conn, self.h2_conn.data_to_send())
         else:

mitmproxy/proxy/layers/http/_http_h2.py

@@ -6,6 +6,7 @@ import h2.connection
 import h2.events
 import h2.exceptions
 import h2.settings
+import h2.stream
 
 
 class H2ConnectionLogger(h2.config.DummyLogger):
@@ -38,11 +39,11 @@ class BufferedH2Connection(h2.connection.H2Connection):
         self.stream_buffers = collections.defaultdict(collections.deque)
 
     def send_data(
-            self,
+        self,
-            stream_id: int,
+        stream_id: int,
-            data: bytes,
+        data: bytes,
-            end_stream: bool = False,
+        end_stream: bool = False,
-            pad_length: None = None
+        pad_length: None = None
     ) -> None:
         """
         Send data on a given stream.
@@ -79,6 +80,13 @@ class BufferedH2Connection(h2.connection.H2Connection):
                     SendH2Data(data, end_stream)
                 )
 
+    def end_stream(self, stream_id) -> None:
+        self.send_data(stream_id, b"", end_stream=True)
+
+    def reset_stream(self, stream_id: int, error_code: int = 0) -> None:
+        self.stream_buffers.pop(stream_id, None)
+        super().reset_stream(stream_id, error_code)
+
     def receive_data(self, data: bytes):
         events = super().receive_data(data)
         ret = []
@@ -103,6 +111,12 @@ class BufferedH2Connection(h2.connection.H2Connection):
         """
         The window for a specific stream has updated. Send as much buffered data as possible.
         """
+        # If the stream has been reset in the meantime, we just clear the buffer.
+        stream: h2.stream.H2Stream = self.streams[stream_id]
+        if stream.state_machine.state not in (h2.stream.StreamState.OPEN, h2.stream.StreamState.HALF_CLOSED_REMOTE):
+            self.stream_buffers.pop(stream_id, None)
+            return False
+
         available_window = self.local_flow_control_window(stream_id)
         sent_any_data = False
         while available_window > 0 and stream_id in self.stream_buffers:
@@ -129,7 +143,7 @@ class BufferedH2Connection(h2.connection.H2Connection):
 
         return sent_any_data
 
-    def connection_window_updated(self):
+    def connection_window_updated(self) -> None:
         """
         The connection window has updated. Send data from buffers in a round-robin fashion.
         """

test/mitmproxy/proxy/layers/http/test_http2.py

@@ -282,9 +282,7 @@ def test_no_normalization(tctx):
             >> reply()
             << http.HttpResponseHook(flow)
             >> reply()
-            << SendData(tctx.client,
+            << SendData(tctx.client, cff.build_headers_frame(response_headers, flags=["END_STREAM"]).serialize())
-                        cff.build_headers_frame(response_headers).serialize() +
-                        cff.build_data_frame(b"", flags=["END_STREAM"]).serialize())
     )
     assert flow().request.headers.fields == ((b"Should-Not-Be-Capitalized! ", b" :) "),)
     assert flow().response.headers.fields == ((b"Same", b"Here"),)