From 91d4d474064cdb822a0308175c8a435ecb83a733 Mon Sep 17 00:00:00 2001 From: adarshx01 Date: Mon, 17 Aug 2026 18:30:43 +0530 Subject: [PATCH 1/2] Ignore leftover frames received after the connection is closed. --- CHANGELOG.rst | 8 ++++++++ src/h2/connection.py | 7 +++++++ tests/test_basic_logic.py | 25 +++++++++++++++++++++++++ 3 files changed, 40 insertions(+) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 44134f898..3afe6f3b4 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -1,6 +1,14 @@ Release History =============== +4.4.2 (unreleased) +------------------ + +**Bugfixes** + +- Ignore leftover frames received after the connection has closed, instead of + raising ``ProtocolError``. Further GOAWAY frames are still processed. + 4.4.1 (2026-08-03) ------------------ diff --git a/src/h2/connection.py b/src/h2/connection.py index e5059a21c..6db29d705 100644 --- a/src/h2/connection.py +++ b/src/h2/connection.py @@ -1541,6 +1541,13 @@ def _receive_frame(self, frame: Frame) -> list[Event]: """ events: list[Event] self.config.logger.trace("Received frame: %s", repr(frame)) + # Leftover frames can still be in the TCP buffer after we closed. + # Extra GOAWAYs are already valid; anything else is not a protocol error. + if ( + self.state_machine.state == ConnectionState.CLOSED + and not isinstance(frame, GoAwayFrame) + ): + return [] try: # I don't love using __class__ here, maybe reconsider it. frames, events = self._frame_dispatch_table[frame.__class__](frame) diff --git a/tests/test_basic_logic.py b/tests/test_basic_logic.py index d1bc0f1eb..d286aaab5 100644 --- a/tests/test_basic_logic.py +++ b/tests/test_basic_logic.py @@ -1788,6 +1788,31 @@ def test_receiving_goaway_frame(self, frame_factory) -> None: assert not c.data_to_send() + def test_frames_after_goaway_are_ignored(self, frame_factory) -> None: + """ + Frames already in the receive buffer after GOAWAY must not raise + ProtocolError. Further GOAWAY frames are still accepted. + """ + c = h2.connection.H2Connection(config=self.server_config) + c.initiate_connection() + c.receive_data(frame_factory.preamble()) + c.clear_outbound_data_buffer() + + goaway = frame_factory.build_goaway_frame(last_stream_id=0) + events = c.receive_data(goaway.serialize()) + assert len(events) == 1 + assert isinstance(events[0], h2.events.ConnectionTerminated) + + ping = frame_factory.build_ping_frame(b"\x01\x02\x03\x04\x05\x06\x07\x08") + events = c.receive_data(ping.serialize()) + assert events == [] + assert not c.data_to_send() + assert c.state_machine.state == h2.connection.ConnectionState.CLOSED + + events = c.receive_data(goaway.serialize()) + assert len(events) == 1 + assert isinstance(events[0], h2.events.ConnectionTerminated) + def test_receiving_multiple_goaway_frames(self, frame_factory) -> None: """ Multiple GOAWAY frames can be received at once, and are allowed. Each From 448f065d099d9cdec712d5c828938ac0ad14067a Mon Sep 17 00:00:00 2001 From: Benjamin Bolton <7146063+benpbolton@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:12:01 -0700 Subject: [PATCH 2/2] Drain open streams after a graceful GOAWAY RFC 9113, Section 6.8 lets streams at or below a GOAWAY's last stream ID complete. Servers often shut down in two stages: GOAWAY(NO_ERROR) with a high last stream ID, a PING, then a GOAWAY with the real last stream ID. h2 moved to CLOSED on any GOAWAY, so the PING or the rest of an open response raised "Invalid input ... in state ConnectionState.CLOSED". After a GOAWAY with NO_ERROR, the connection now drains: - The state stays CLOSED, so consumers that check for CLOSED still stop opening streams on the connection. H2ConnectionStateMachine.draining reports the drain. - Frames for open streams are still processed. PING and SETTINGS are still acknowledged. PUSH_PROMISE can still be received. The outbound buffer is kept. - The drain keeps a shadow of the open state, so the client and server rules of the open connection still apply. - New local streams (send_headers on a new stream ID, push_stream, initiate_upgrade_connection) raise ProtocolError before any state change, and the drain continues. - A further graceful GOAWAY continues the drain. A GOAWAY with an error code, an invalid input or close_connection() ends it. A graceful GOAWAY never reopens a connection that is already closed. ConnectionTerminated is still fired for every GOAWAY. h2 does not reset or filter streams above last_stream_id; the application decides. Fixes #1199. Fixes the receive side of #1181. Builds on #1324. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.rst | 14 + src/h2/connection.py | 99 ++++++- src/h2/events.py | 18 +- tests/test_basic_logic.py | 10 +- tests/test_graceful_goaway.py | 489 ++++++++++++++++++++++++++++++++++ 5 files changed, 623 insertions(+), 7 deletions(-) create mode 100644 tests/test_graceful_goaway.py diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 3afe6f3b4..0fbfd6c68 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -4,6 +4,20 @@ Release History 4.4.2 (unreleased) ------------------ +**API Changes (Backward Compatible)** + +- Support graceful shutdown by the remote peer (RFC 9113, Section 6.8). After + a GOAWAY with error code ``NO_ERROR``, the connection drains: its state is + ``CLOSED`` and no new local streams may be opened, but frames for open + streams are still processed, PING and SETTINGS frames are still + acknowledged, and the outbound data buffer is kept. The client and server + rules of the open connection still apply. A further graceful GOAWAY keeps + draining; a GOAWAY with an error code, an invalid input or + ``close_connection()`` ends the drain, and a connection that is already + closed is never reopened by a later GOAWAY. ``H2Connection.state_machine.draining`` + reports the drain. ``ConnectionTerminated`` is still fired for every GOAWAY. + (#1181, #1199) + **Bugfixes** - Ignore leftover frames received after the connection has closed, instead of diff --git a/src/h2/connection.py b/src/h2/connection.py index 6db29d705..a2e5aadc2 100644 --- a/src/h2/connection.py +++ b/src/h2/connection.py @@ -247,8 +247,58 @@ class H2ConnectionStateMachine: (None, ConnectionState.CLOSED), } + # Inputs that stay valid in the CLOSED state while the connection drains + # after the remote peer sent a graceful GOAWAY (error code NO_ERROR). + # RFC 9113, Section 6.8: streams that were already open may still + # complete, but the receiver of the GOAWAY must not open new streams. + # SEND_PUSH_PROMISE is therefore absent; H2Connection refuses other new + # local streams, since it knows the stream IDs. An input is accepted only + # if the open connection would also accept it (see _draining_state). + _draining_inputs = frozenset({ + ConnectionInputs.SEND_HEADERS, + ConnectionInputs.SEND_DATA, + ConnectionInputs.SEND_WINDOW_UPDATE, + ConnectionInputs.SEND_PING, + ConnectionInputs.SEND_SETTINGS, + ConnectionInputs.SEND_RST_STREAM, + ConnectionInputs.SEND_PRIORITY, + ConnectionInputs.SEND_ALTERNATIVE_SERVICE, + ConnectionInputs.RECV_HEADERS, + ConnectionInputs.RECV_PUSH_PROMISE, + ConnectionInputs.RECV_DATA, + ConnectionInputs.RECV_WINDOW_UPDATE, + ConnectionInputs.RECV_PING, + ConnectionInputs.RECV_SETTINGS, + ConnectionInputs.RECV_RST_STREAM, + ConnectionInputs.RECV_PRIORITY, + ConnectionInputs.RECV_ALTERNATIVE_SERVICE, + }) + def __init__(self) -> None: self.state = ConnectionState.IDLE + # While draining: the state the connection would be in had the GOAWAY + # not arrived (IDLE, CLIENT_OPEN or SERVER_OPEN). It keeps the normal + # client and server rules in force. None when not draining. + self._draining_state: ConnectionState | None = None + + @property + def draining(self) -> bool: + """ + Whether the connection is draining: the remote peer sent a graceful + GOAWAY, so the state is CLOSED and this side may not open new streams, + but frames for open streams are still processed. The peer that sent + the GOAWAY may still open streams (RFC 9113, Section 6.8). + """ + return self._draining_state is not None + + def _start_draining(self, open_state: ConnectionState) -> None: + """ + Start (or continue) draining after a graceful GOAWAY from the peer. + ``open_state`` is the state before the GOAWAY. It must not be CLOSED: + a connection that is already closed stays closed. + """ + assert open_state is not ConnectionState.CLOSED + self._draining_state = open_state def process_input(self, input_: ConnectionInputs) -> list[Event]: """ @@ -258,6 +308,16 @@ def process_input(self, input_: ConnectionInputs) -> list[Event]: msg = "Input must be an instance of ConnectionInputs" raise ValueError(msg) # noqa: TRY004 + if self._draining_state is not None: + transition = self._transitions.get((self._draining_state, input_)) + if input_ in self._draining_inputs and transition is not None: + self._draining_state = transition[1] + return [] + # Anything else (a GOAWAY, or an input the open connection would + # also reject) ends the drain. A further graceful GOAWAY from the + # peer continues it (see H2Connection._receive_goaway_frame). + self._draining_state = None + try: func, target_state = self._transitions[(self.state, input_)] except KeyError as e: @@ -573,6 +633,8 @@ def initiate_upgrade_connection(self, settings_header: bytes | None = None) -> b "Upgrade connection. Current settings: %s", self.local_settings, ) + self._refuse_new_stream_while_draining(1) + frame_data = None # Begin by getting the preamble in place. self.initiate_connection() @@ -609,6 +671,19 @@ def initiate_upgrade_connection(self, settings_header: bytes | None = None) -> b self.streams[1].upgrade(self.config.client_side) return frame_data + def _refuse_new_stream_while_draining(self, stream_id: int) -> None: + """ + The receiver of a GOAWAY must not open new streams (RFC 9113, Section + 6.8). Raise before any state changes, so the drain continues for the + streams that are already open. + """ + if self.state_machine.draining: + msg = ( + f"Cannot open stream {stream_id}: the remote peer sent GOAWAY, " + "so no new streams may be opened" + ) + raise ProtocolError(msg) + def _get_or_create_stream(self, stream_id: int, allowed_ids: AllowedStreamIDs) -> H2Stream: """ Gets a stream by its stream ID. Will create one if one does not already @@ -794,6 +869,7 @@ def send_headers(self, # Check we can open the stream. if stream_id not in self.streams: + self._refuse_new_stream_while_draining(stream_id) max_open_streams = self.remote_settings.max_concurrent_streams value = self.open_outbound_streams # take a copy due to the property accessor having side affects if (value + 1) > max_open_streams: @@ -1010,6 +1086,7 @@ def push_stream(self, msg = "Remote peer has disabled stream push" raise ProtocolError(msg) + self._refuse_new_stream_while_draining(promised_stream_id) self.state_machine.process_input(ConnectionInputs.SEND_PUSH_PROMISE) stream = self._get_stream_by_id(stream_id) @@ -1543,8 +1620,10 @@ def _receive_frame(self, frame: Frame) -> list[Event]: self.config.logger.trace("Received frame: %s", repr(frame)) # Leftover frames can still be in the TCP buffer after we closed. # Extra GOAWAYs are already valid; anything else is not a protocol error. + # While draining after a graceful GOAWAY, frames are still processed. if ( self.state_machine.state == ConnectionState.CLOSED + and not self.state_machine.draining and not isinstance(frame, GoAwayFrame) ): return [] @@ -1901,12 +1980,28 @@ def _receive_goaway_frame(self, frame: GoAwayFrame) -> tuple[list[Frame], list[E """ Receive a GOAWAY frame on the connection. """ + # The state the connection would be in without this GOAWAY: the state + # before it, or the state an existing drain keeps. None if the + # connection is already closed for good (an error GOAWAY, a protocol + # error, or close_connection()); a later graceful GOAWAY must not + # revive it (RFC 9113, Section 5.4.1). + open_state: ConnectionState | None = self.state_machine._draining_state + if open_state is None and self.state_machine.state is not ConnectionState.CLOSED: + open_state = self.state_machine.state + events = self.state_machine.process_input( ConnectionInputs.RECV_GOAWAY, ) - # Clear the outbound data buffer: we cannot send further data now. - self.clear_outbound_data_buffer() + if frame.error_code == ErrorCodes.NO_ERROR and open_state is not None: + # A graceful shutdown (RFC 9113, Section 6.8): no new streams, but + # streams that are already open may complete. Keep the outbound + # buffer, which may hold WINDOW_UPDATE, SETTINGS or PING ACK frames + # those streams need. + self.state_machine._start_draining(open_state) + else: + # Clear the outbound data buffer: we cannot send further data now. + self.clear_outbound_data_buffer() # Fire an appropriate ConnectionTerminated event. new_event = ConnectionTerminated() diff --git a/src/h2/events.py b/src/h2/events.py index e3accea76..b8e41081d 100644 --- a/src/h2/events.py +++ b/src/h2/events.py @@ -562,8 +562,22 @@ def __repr__(self) -> str: class ConnectionTerminated(Event): """ The ConnectionTerminated event is fired when a connection is torn down by - the remote peer using a GOAWAY frame. Once received, no further action may - be taken on the connection: a new connection must be established. + the remote peer using a GOAWAY frame. Once received, no new streams may be + opened on the connection: a new connection must be established for them. + + If ``error_code`` is ``NO_ERROR``, this is a graceful shutdown (RFC 9113, + Section 6.8): streams up to ``last_stream_id`` may still complete. On a + connection that was not already closed, h2 starts (or continues) a drain + and keeps processing frames for open streams; check + ``H2Connection.state_machine.draining`` for the current state. Locally initiated + streams above ``last_stream_id`` were not processed by the peer and are + safe to retry elsewhere. h2 does not reset or filter them, and does not + check that a later GOAWAY keeps ``last_stream_id`` from increasing: the + application decides. With any other error code, no further action may be + taken on the connection. + + .. versionchanged:: 4.4.2 + A graceful GOAWAY no longer prevents open streams from completing. """ def __init__(self) -> None: diff --git a/tests/test_basic_logic.py b/tests/test_basic_logic.py index d286aaab5..0bdf95ecd 100644 --- a/tests/test_basic_logic.py +++ b/tests/test_basic_logic.py @@ -1790,15 +1790,19 @@ def test_receiving_goaway_frame(self, frame_factory) -> None: def test_frames_after_goaway_are_ignored(self, frame_factory) -> None: """ - Frames already in the receive buffer after GOAWAY must not raise - ProtocolError. Further GOAWAY frames are still accepted. + Frames already in the receive buffer after a GOAWAY with an error code + must not raise ProtocolError. Further GOAWAY frames are still accepted. + (After a graceful GOAWAY the connection drains instead; see + test_graceful_goaway.py.) """ c = h2.connection.H2Connection(config=self.server_config) c.initiate_connection() c.receive_data(frame_factory.preamble()) c.clear_outbound_data_buffer() - goaway = frame_factory.build_goaway_frame(last_stream_id=0) + goaway = frame_factory.build_goaway_frame( + last_stream_id=0, error_code=h2.errors.ErrorCodes.PROTOCOL_ERROR, + ) events = c.receive_data(goaway.serialize()) assert len(events) == 1 assert isinstance(events[0], h2.events.ConnectionTerminated) diff --git a/tests/test_graceful_goaway.py b/tests/test_graceful_goaway.py new file mode 100644 index 000000000..dffbd6381 --- /dev/null +++ b/tests/test_graceful_goaway.py @@ -0,0 +1,489 @@ +""" +Tests for graceful shutdown: a GOAWAY frame with NO_ERROR. + +RFC 9113, Section 6.8: the receiver of a GOAWAY must not open new streams, but +"activity on streams numbered lower than or equal to the last stream identifier +might still complete successfully". Servers commonly send GOAWAY(NO_ERROR, +2**31-1), then a PING, then a second GOAWAY with the real last stream ID. +""" +from __future__ import annotations + +import pytest + +import h2.config +import h2.connection +import h2.errors +import h2.events +import h2.exceptions + +MAX_STREAM_ID = 2**31 - 1 + + +class TestGracefulGoAwayClient: + """ + A client receives a graceful GOAWAY while a request is in flight. + """ + + request_headers = [ + (":authority", "example.com"), + (":path", "/"), + (":scheme", "https"), + (":method", "GET"), + ] + post_headers = [ + (":authority", "example.com"), + (":path", "/upload"), + (":scheme", "https"), + (":method", "POST"), + ] + response_headers = [ + (":status", "200"), + ("server", "fake-serv/0.1.0"), + ] + + def client_with_request( + self, frame_factory, *, end_stream=True, + ) -> h2.connection.H2Connection: + c = h2.connection.H2Connection() + c.initiate_connection() + c.send_headers( + 1, self.request_headers if end_stream else self.post_headers, + end_stream=end_stream, + ) + c.receive_data(frame_factory.build_settings_frame({}).serialize()) + c.clear_outbound_data_buffer() + return c + + def test_ping_after_graceful_goaway_is_acked(self, frame_factory) -> None: + """ + GOAWAY(NO_ERROR) followed by PING in the same read does not raise, and + the PING is acknowledged (this is the two-stage shutdown recommended + by RFC 9113). + """ + c = self.client_with_request(frame_factory) + goaway = frame_factory.build_goaway_frame(last_stream_id=MAX_STREAM_ID) + ping = frame_factory.build_ping_frame(b"\x01" * 8) + + events = c.receive_data(goaway.serialize() + ping.serialize()) + + assert [type(e) for e in events] == [ + h2.events.ConnectionTerminated, h2.events.PingReceived, + ] + assert events[0].error_code == h2.errors.ErrorCodes.NO_ERROR + assert events[0].last_stream_id == MAX_STREAM_ID + assert c.state_machine.state == h2.connection.ConnectionState.CLOSED + ack = frame_factory.build_ping_frame(b"\x01" * 8, flags=["ACK"]) + assert c.data_to_send() == ack.serialize() + + def test_in_flight_response_completes(self, frame_factory) -> None: + """ + A response on a stream at or below last_stream_id is delivered in full + after a graceful GOAWAY. + """ + c = self.client_with_request(frame_factory) + goaway = frame_factory.build_goaway_frame(last_stream_id=1) + headers = frame_factory.build_headers_frame(self.response_headers) + data = frame_factory.build_data_frame(b"hello", flags=["END_STREAM"]) + + events = c.receive_data( + goaway.serialize() + headers.serialize() + data.serialize(), + ) + + assert [type(e) for e in events] == [ + h2.events.ConnectionTerminated, + h2.events.ResponseReceived, + h2.events.DataReceived, + h2.events.StreamEnded, + ] + assert events[2].data == b"hello" + + def test_outbound_data_is_kept_and_flow_control_continues(self, frame_factory) -> None: + """ + A graceful GOAWAY does not discard queued outbound frames, and the + client can still acknowledge received data with WINDOW_UPDATE frames. + """ + c = self.client_with_request(frame_factory) + c.ping(b"\x02" * 8) + queued = c.data_to_send() + c.ping(b"\x02" * 8) + + goaway = frame_factory.build_goaway_frame(last_stream_id=1) + headers = frame_factory.build_headers_frame(self.response_headers) + data = frame_factory.build_data_frame(b"x" * 1000) + events = c.receive_data( + goaway.serialize() + headers.serialize() + data.serialize(), + ) + assert c.data_to_send() == queued + + assert events[-1].flow_controlled_length == 1000 + c.increment_flow_control_window(1000, stream_id=1) + c.increment_flow_control_window(1000) + assert c.data_to_send() == ( + frame_factory.build_window_update_frame( + stream_id=1, increment=1000, + ).serialize() + + frame_factory.build_window_update_frame( + stream_id=0, increment=1000, + ).serialize() + ) + + def test_client_can_finish_request_body(self, frame_factory) -> None: + """ + A client can keep sending the body of an in-flight request after a + graceful GOAWAY. + """ + c = self.client_with_request(frame_factory, end_stream=False) + goaway = frame_factory.build_goaway_frame(last_stream_id=1) + c.receive_data(goaway.serialize()) + + c.send_data(1, b"body", end_stream=True) + + expected = frame_factory.build_data_frame(b"body", flags=["END_STREAM"]) + assert c.data_to_send() == expected.serialize() + + def test_cannot_open_new_stream_while_draining(self, frame_factory) -> None: + """ + Opening a new stream after a graceful GOAWAY raises ProtocolError, but + the in-flight stream still completes afterwards. + """ + c = self.client_with_request(frame_factory) + goaway = frame_factory.build_goaway_frame(last_stream_id=1) + c.receive_data(goaway.serialize()) + + with pytest.raises(h2.exceptions.ProtocolError): + c.send_headers(3, self.request_headers, end_stream=True) + assert not c.data_to_send() + + headers = frame_factory.build_headers_frame( + self.response_headers, flags=["END_STREAM"], + ) + events = c.receive_data(headers.serialize()) + assert [type(e) for e in events] == [ + h2.events.ResponseReceived, h2.events.StreamEnded, + ] + + def test_second_graceful_goaway_keeps_draining(self, frame_factory) -> None: + """ + The two-stage shutdown (GOAWAY with 2**31-1, PING, GOAWAY with the + real last stream ID) keeps the in-flight stream usable. + """ + c = self.client_with_request(frame_factory) + first = frame_factory.build_goaway_frame(last_stream_id=MAX_STREAM_ID) + ping = frame_factory.build_ping_frame(b"\x03" * 8) + second = frame_factory.build_goaway_frame(last_stream_id=1) + headers = frame_factory.build_headers_frame( + self.response_headers, flags=["END_STREAM"], + ) + + events = c.receive_data( + first.serialize() + ping.serialize() + second.serialize() + + headers.serialize(), + ) + + assert [type(e) for e in events] == [ + h2.events.ConnectionTerminated, + h2.events.PingReceived, + h2.events.ConnectionTerminated, + h2.events.ResponseReceived, + h2.events.StreamEnded, + ] + assert events[2].last_stream_id == 1 + + def test_reset_after_graceful_goaway(self, frame_factory) -> None: + """ + The frame sequence from mitmproxy/mitmproxy#5897: the response ends, + the server resets the stream, sends a graceful GOAWAY and resets the + stream again. None of this is a connection error. + """ + c = self.client_with_request(frame_factory) + frames = [ + frame_factory.build_headers_frame(self.response_headers), + frame_factory.build_data_frame(b""), + frame_factory.build_data_frame(b"", flags=["END_STREAM"]), + frame_factory.build_rst_stream_frame(stream_id=1, error_code=0), + frame_factory.build_goaway_frame(last_stream_id=MAX_STREAM_ID), + frame_factory.build_rst_stream_frame(stream_id=1, error_code=5), + ] + + events = c.receive_data(b"".join(f.serialize() for f in frames)) + + assert isinstance(events[-1], h2.events.ConnectionTerminated) + assert events[-1].error_code == h2.errors.ErrorCodes.NO_ERROR + assert c.state_machine.state == h2.connection.ConnectionState.CLOSED + + def test_settings_after_graceful_goaway_are_acked(self, frame_factory) -> None: + """ + SETTINGS frames received while draining are applied and acknowledged. + """ + c = self.client_with_request(frame_factory) + goaway = frame_factory.build_goaway_frame(last_stream_id=1) + settings = frame_factory.build_settings_frame( + {h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: 5}, + ) + + events = c.receive_data(goaway.serialize() + settings.serialize()) + + assert isinstance(events[-1], h2.events.RemoteSettingsChanged) + assert c.data_to_send() == frame_factory.build_settings_frame( + {}, ack=True, + ).serialize() + + def test_error_goaway_ends_draining(self, frame_factory) -> None: + """ + A GOAWAY with an error code after a graceful one closes the connection + for good: the outbound buffer is cleared and later frames are ignored. + """ + c = self.client_with_request(frame_factory) + graceful = frame_factory.build_goaway_frame(last_stream_id=1) + error = frame_factory.build_goaway_frame( + last_stream_id=1, error_code=h2.errors.ErrorCodes.INTERNAL_ERROR, + ) + c.receive_data(graceful.serialize()) + c.ping(b"\x04" * 8) + + events = c.receive_data(error.serialize()) + assert len(events) == 1 + assert events[0].error_code == h2.errors.ErrorCodes.INTERNAL_ERROR + assert not c.data_to_send() + + headers = frame_factory.build_headers_frame( + self.response_headers, flags=["END_STREAM"], + ) + assert c.receive_data(headers.serialize()) == [] + + def test_local_close_ends_draining(self, frame_factory) -> None: + """ + Calling close_connection() while draining sends GOAWAY and closes the + connection for good; later frames are ignored. + """ + c = self.client_with_request(frame_factory) + c.receive_data(frame_factory.build_goaway_frame(last_stream_id=1).serialize()) + + c.close_connection() + assert c.data_to_send() == frame_factory.build_goaway_frame( + last_stream_id=0, + ).serialize() + + ping = frame_factory.build_ping_frame(b"\x05" * 8) + assert c.receive_data(ping.serialize()) == [] + assert not c.data_to_send() + + def test_non_graceful_goaway_clears_outbound_data(self, frame_factory) -> None: + """ + A GOAWAY with an error code keeps its existing behaviour: the outbound + buffer is cleared and in-flight streams get no further frames. + """ + c = self.client_with_request(frame_factory) + c.ping(b"\x06" * 8) + goaway = frame_factory.build_goaway_frame( + last_stream_id=1, error_code=h2.errors.ErrorCodes.PROTOCOL_ERROR, + ) + headers = frame_factory.build_headers_frame( + self.response_headers, flags=["END_STREAM"], + ) + + events = c.receive_data(goaway.serialize() + headers.serialize()) + + assert [type(e) for e in events] == [h2.events.ConnectionTerminated] + assert not c.data_to_send() + + +class TestGracefulGoAwayServer: + """ + A server receives a graceful GOAWAY from the client while it is still + answering a request. + """ + + request_headers = [ + (":authority", "example.com"), + (":path", "/"), + (":scheme", "https"), + (":method", "GET"), + ] + response_headers = [ + (":status", "200"), + ("server", "fake-serv/0.1.0"), + ] + server_config = h2.config.H2Configuration(client_side=False) + + def test_server_can_finish_response(self, frame_factory) -> None: + c = h2.connection.H2Connection(config=self.server_config) + c.initiate_connection() + c.receive_data(frame_factory.preamble()) + c.receive_data(frame_factory.build_headers_frame( + self.request_headers, flags=["END_STREAM"], + ).serialize()) + c.clear_outbound_data_buffer() + + events = c.receive_data( + frame_factory.build_goaway_frame(last_stream_id=0).serialize(), + ) + assert isinstance(events[0], h2.events.ConnectionTerminated) + + c.send_headers(1, self.response_headers) + c.send_data(1, b"done", end_stream=True) + + frame_factory.refresh_encoder() + expected = ( + frame_factory.build_headers_frame(self.response_headers).serialize() + + frame_factory.build_data_frame(b"done", flags=["END_STREAM"]).serialize() + ) + assert c.data_to_send() == expected + + +class TestGracefulGoAwayEdges: + """ + A graceful GOAWAY must not revive a closed connection, must not let local + code open streams by other routes, and must keep the client and server + rules of the open connection. + """ + + request_headers = TestGracefulGoAwayClient.request_headers + response_headers = TestGracefulGoAwayClient.response_headers + server_config = h2.config.H2Configuration(client_side=False) + + def client_with_request(self, frame_factory) -> h2.connection.H2Connection: + return TestGracefulGoAwayClient().client_with_request(frame_factory) + + def assert_terminal(self, c, frame_factory) -> None: + graceful = frame_factory.build_goaway_frame(last_stream_id=1) + ping = frame_factory.build_ping_frame(b"\x07" * 8) + events = c.receive_data(graceful.serialize() + ping.serialize()) + assert [type(e) for e in events] == [h2.events.ConnectionTerminated] + assert not c.state_machine.draining + assert not c.data_to_send() + + def test_graceful_goaway_after_local_close_stays_closed(self, frame_factory) -> None: + c = self.client_with_request(frame_factory) + c.close_connection() + c.clear_outbound_data_buffer() + self.assert_terminal(c, frame_factory) + + def test_graceful_goaway_after_error_goaway_stays_closed(self, frame_factory) -> None: + c = self.client_with_request(frame_factory) + c.receive_data(frame_factory.build_goaway_frame( + last_stream_id=1, error_code=h2.errors.ErrorCodes.INTERNAL_ERROR, + ).serialize()) + self.assert_terminal(c, frame_factory) + + def test_graceful_goaway_after_protocol_error_stays_closed(self, frame_factory) -> None: + c = self.client_with_request(frame_factory) + # DATA on an idle stream is a connection error. + with pytest.raises(h2.exceptions.ProtocolError): + c.receive_data(frame_factory.build_data_frame(b"x", stream_id=7).serialize()) + c.clear_outbound_data_buffer() + self.assert_terminal(c, frame_factory) + + def test_push_stream_refused_while_draining(self, frame_factory) -> None: + """ + A server that received a graceful GOAWAY cannot push, and the refusal + does not end the drain: the response can still be sent. + """ + c = h2.connection.H2Connection(config=self.server_config) + c.initiate_connection() + c.receive_data(frame_factory.preamble()) + c.receive_data(frame_factory.build_headers_frame( + self.request_headers, flags=["END_STREAM"], + ).serialize()) + c.receive_data(frame_factory.build_goaway_frame(last_stream_id=0).serialize()) + c.clear_outbound_data_buffer() + + with pytest.raises(h2.exceptions.ProtocolError): + c.push_stream(1, 2, self.request_headers) + assert c.state_machine.draining + assert not c.data_to_send() + + c.send_headers(1, self.response_headers, end_stream=True) + assert c.data_to_send() + + def test_upgrade_refused_while_draining(self, frame_factory) -> None: + c = h2.connection.H2Connection() + c.receive_data(frame_factory.build_goaway_frame(last_stream_id=0).serialize()) + assert c.state_machine.draining + + with pytest.raises(h2.exceptions.ProtocolError): + c.initiate_upgrade_connection() + assert not c.data_to_send() + assert 1 not in c.streams + + def test_push_promise_received_while_draining(self, frame_factory) -> None: + """ + The sender of a GOAWAY may still push on an open request; only the + receiver may not open streams. + """ + c = self.client_with_request(frame_factory) + frames = [ + frame_factory.build_goaway_frame(last_stream_id=1), + frame_factory.build_push_promise_frame( + stream_id=1, promised_stream_id=2, headers=self.request_headers, + flags=["END_HEADERS"], + ), + frame_factory.build_headers_frame( + self.response_headers, flags=["END_STREAM"], + ), + ] + + events = c.receive_data(b"".join(f.serialize() for f in frames)) + + assert [type(e) for e in events] == [ + h2.events.ConnectionTerminated, + h2.events.PushedStreamReceived, + h2.events.ResponseReceived, + h2.events.StreamEnded, + ] + assert c.state_machine.draining + + def test_role_rules_still_apply_while_draining(self, frame_factory) -> None: + """ + A client may not send ALTSVC frames. Draining does not change that. + """ + c = self.client_with_request(frame_factory) + c.receive_data(frame_factory.build_goaway_frame(last_stream_id=1).serialize()) + + with pytest.raises(h2.exceptions.ProtocolError): + c.advertise_alternative_service(b'h2=":8000"', origin=b"example.com") + assert not c.data_to_send() + + def test_server_rules_still_apply_while_draining(self, frame_factory) -> None: + """ + A server may not receive PUSH_PROMISE, draining or not. + """ + c = h2.connection.H2Connection(config=self.server_config) + c.initiate_connection() + c.receive_data(frame_factory.preamble()) + c.receive_data(frame_factory.build_headers_frame(self.request_headers).serialize()) + c.receive_data(frame_factory.build_goaway_frame(last_stream_id=0).serialize()) + + with pytest.raises(h2.exceptions.ProtocolError): + c.receive_data(frame_factory.build_push_promise_frame( + stream_id=1, promised_stream_id=2, headers=self.request_headers, + flags=["END_HEADERS"], + ).serialize()) + assert not c.state_machine.draining + + def test_peer_stream_after_graceful_goaway_from_idle(self, frame_factory) -> None: + """ + A server that receives a graceful GOAWAY before any stream may still + serve a stream the client opens afterwards: only the GOAWAY receiver + is barred from opening streams. The drain tracks the open state, so + the response can be sent. + """ + c = h2.connection.H2Connection(config=self.server_config) + c.initiate_connection() + c.receive_data(frame_factory.preamble()) + c.receive_data(frame_factory.build_goaway_frame(last_stream_id=0).serialize()) + assert c.state_machine.draining + + events = c.receive_data( + frame_factory.build_headers_frame(self.request_headers).serialize() + + frame_factory.build_data_frame(b"x", flags=["END_STREAM"]).serialize(), + ) + assert [type(e) for e in events] == [ + h2.events.RequestReceived, h2.events.DataReceived, h2.events.StreamEnded, + ] + c.clear_outbound_data_buffer() + + c.send_headers(1, self.response_headers) + c.send_data(1, b"ok", end_stream=True) + assert c.data_to_send() + assert c.state_machine.draining