From 43edffdec4047734f4281feca456771abe99e859 Mon Sep 17 00:00:00 2001 From: antoinetoussaint-byte <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Sun, 25 Oct 2026 22:26:36 -0400 Subject: [PATCH 1/2] fix(connect): the minors of #111 and #117's reviews #111 (review r106): - m1: the channel's own receive bound is one byte past both answer bounds, so a refusal a service passes on from a client whose bound equals the face's is never taken for the face's own (grpcio names the client's bound). - m2/m3: the docstring says a list of enums costs with the length of their names (measured 130x at 22 characters, 373x at 100, 687x at 200), so the JSON default's worst case depends on the service's enum names, and gives the measured process memory of a proto answer (54 to 98 MiB for 15 MiB). #117 (review r107): - m1: the client's SETTINGS, which completes its preface, goes to the server before the face waits for the server's, so a server that waits for the client's first is not stalled. - m2: a server that sends no SETTINGS within read_timeout gets the client a GOAWAY (INTERNAL_ERROR) naming why. - m3: a SETTINGS frame is scanned whole as it streams, holding one entry, so the server's limit is found wherever it is in the frame (not only in its first 256 bytes). A first frame is held to the 16 KiB any peer may send before it has the other's SETTINGS. Tests (each fails on main): test_a_refusal_naming_the_faces_own_bound_is_still_the_services, test_a_server_that_waits_for_the_clients_settings_is_not_stalled, test_a_server_that_sends_no_settings_gets_the_client_a_goaway, test_the_servers_limit_is_found_anywhere_in_its_settings (also fed one byte at a time). Co-Authored-By: Claude Opus 5.5 --- templates/connect/connect_gen.py | 103 ++++++++++++++++++++------- testdata/connect/test_connect_gen.py | 78 +++++++++++++++++++- 2 files changed, 154 insertions(+), 27 deletions(-) diff --git a/templates/connect/connect_gen.py b/templates/connect/connect_gen.py index 364549d..27b9f89 100644 --- a/templates/connect/connect_gen.py +++ b/templates/connect/connect_gen.py @@ -23,16 +23,19 @@ - ``max_response_bytes`` for an answer relayed as proto (Connect and gRPC-Web proto): by default the larger of ``DEFAULT_MAX_RESPONSE_BYTES`` (16 MiB) and ``max_message_bytes``. - The face holds such an answer about twice over while it writes it, beyond the - service's own copy: measured, 33 MiB for a 16 MiB answer. + The face holds such an answer more than once while it writes it: measured as process + memory, 54 to 98 MiB for a 15 MiB answer, by protocol. - ``max_json_response_bytes`` for an answer converted to JSON: by default the larger of ``DEFAULT_MAX_JSON_RESPONSE_BYTES`` (4 MiB, the bound every answer had before the two were apart) and ``max_message_bytes``. It is checked before the conversion, because the conversion is what costs: measured as process memory (protobuf 6.33, one 4 MiB answer), - a packed list of floats peaks at 18 times the proto, nested messages at 62 times, and a - list of enums at 130 times (its JSON alone is 25 times the proto). So up to about - 520 MiB at the default, for one conversion; conversions run on the face's event loop - one at a time, so that peak does not add up across calls. + a packed list of floats peaks at 18 times the proto and nested messages at 62 times. A + list of enums costs with the length of their names, since each one-byte value becomes + its name in JSON: 130 times with 22-character names, 373 times with 100, 687 times with + 200. So the worst case at the default depends on the service's enum names: about + 520 MiB for one answer with names of about 20 characters, 2.7 GiB with 200. + Conversions run on the face's event loop one at a time, so that peak does not add up + across calls; a service with long enum names in large lists lowers the bound. Past its bound a call ends ``failed_precondition`` naming the bound: the answer is too large for this face's configuration, and asking again does not change it. Within it, @@ -93,7 +96,9 @@ _END_STREAM_FLAG, _ACK_FLAG, _END_HEADERS_FLAG = 0x1, 0x1, 0x4 _MAX_CONCURRENT_STREAMS = 0x3 _PUSH_PROMISE = 0x5 -_PROTOCOL_ERROR, _CANCEL, _ENHANCE_YOUR_CALM = 0x1, 0x8, 0xB +_PROTOCOL_ERROR, _INTERNAL_ERROR, _CANCEL, _ENHANCE_YOUR_CALM = 0x1, 0x2, 0x8, 0xB +# The largest frame a peer may send before it has the other's SETTINGS (RFC 9113 4.2). +_INITIAL_MAX_FRAME = 16384 # Calls one relayed connection may hold open when the gRPC server advertises no # SETTINGS_MAX_CONCURRENT_STREAMS (grpcio advertises none unless the service sets # grpc.max_concurrent_streams, as the scaffold's server does). The face advertises it to @@ -113,7 +118,8 @@ # How long a closing relay waits for the server's frame in flight to complete, so the # client's GOAWAY lands between two frames. _GOAWAY_GRACE_S = 5.0 -# The most of a SETTINGS or GOAWAY payload the relay reads; the rest is skipped. +# The most of a GOAWAY payload the relay reads (the last stream, the code, the start of +# the reason); the rest is skipped. A SETTINGS frame is scanned whole, keeping nothing. _CONTROL_PAYLOAD = 256 _ENVELOPE = struct.Struct(">BI") _COMPRESSED, _END_STREAM, _TRAILERS = 0x01, 0x02, 0x80 @@ -285,11 +291,14 @@ def _positive(name: str, value: int | None, default: int, message_bound: int) -> class _AnswerBounds: """The answers the face takes from the service: ``proto`` for a codec that relays - them as they are, ``json`` for one that converts them, checked before converting.""" + them as they are, ``json`` for one that converts them, checked before converting. + ``received`` is the channel's own receive bound: one byte past both, so the face + checks every bound itself, and the number grpcio names when it refuses past it is + one no service's client has by chance with the same text.""" def __init__(self, proto: int, json: int): self.proto, self.json = proto, json - self.received = max(proto, json) + self.received = max(proto, json) + 1 def of(self, codec: _Codec) -> tuple[int, str]: if codec.name == "proto": @@ -333,8 +342,10 @@ class _Frames: ``feed`` calls ``on_frame(type, flags, stream, payload)`` for each frame and returns where frames end in the data, each with the bytes ``on_frame`` returned, to be written - right after that frame in this direction. The payload is read for SETTINGS and GOAWAY - only (at most ``_CONTROL_PAYLOAD`` bytes of it) and skipped otherwise. A + right after that frame in this direction. The payload passed is, for a SETTINGS + frame, its last SETTINGS_MAX_CONCURRENT_STREAMS entry if it has one (the frame is + scanned whole as it streams, holding one entry), for a GOAWAY its first + ``_CONTROL_PAYLOAD`` bytes, and empty otherwise. A ``_ProtocolError`` from ``on_frame`` stops the feed; it is kept in ``error`` and the ends before the offending frame are returned.""" @@ -346,6 +357,8 @@ def __init__(self, on_frame): self._payload = bytearray() self._keep = 0 # payload bytes still to read self._skip = 0 # payload bytes still to skip + self._scan = 0 # SETTINGS payload bytes still to scan + self._entry = bytearray() # a SETTINGS entry split across reads self._in_block = False # inside a header block: HEADERS ... CONTINUATION self.error: _ProtocolError | None = None @@ -353,7 +366,7 @@ def __init__(self, on_frame): def at_boundary(self) -> bool: """Between two frames, and outside a header block: where another frame may go (RFC 9113 section 6.10: nothing comes between a HEADERS and its CONTINUATION).""" - return not self._header and not self._keep and not self._skip and not self._in_block + return not (self._header or self._keep or self._skip or self._scan or self._in_block) def feed(self, data: bytes, every_end: bool = False) -> list[tuple[int, bytes]]: """Where frames end in ``data``: each end where ``on_frame`` returned bytes to @@ -379,6 +392,14 @@ def feed(self, data: bytes, every_end: bool = False) -> list[tuple[int, bytes]]: at += step if self._skip: break + elif self._scan: + step = min(self._scan, size - at) + self._scan_settings(data[at : at + step]) + at += step + if self._scan: + break + self._entry.clear() + self._emit() elif self._header or size - at < _FRAME.size: step = min(_FRAME.size - len(self._header), size - at) self._header += data[at : at + step] @@ -390,7 +411,7 @@ def feed(self, data: bytes, every_end: bool = False) -> list[tuple[int, bytes]]: else: # a whole frame header in this chunk: read it in place self._start(*_FRAME.unpack_from(data, at)) at += _FRAME.size - if not (self._keep or self._skip or self._in_block): + if not (self._keep or self._skip or self._scan or self._in_block): clear = at if every_end or self._after: ends.append((at, self._after)) @@ -407,22 +428,37 @@ def _start(self, high, low, kind, flags, stream): self._frame = (kind, flags, stream & 0x7FFFFFFF) if kind in (_HEADERS, _PUSH_PROMISE, _CONTINUATION): self._in_block = not flags & _END_HEADERS_FLAG - self._keep = min(length, _CONTROL_PAYLOAD) if kind in (_SETTINGS, _GOAWAY) else 0 - self._skip = length - self._keep - if not self._keep: + self._keep = min(length, _CONTROL_PAYLOAD) if kind == _GOAWAY else 0 + self._scan = length if kind == _SETTINGS else 0 + self._skip = length - self._keep - self._scan + if not (self._keep or self._scan): self._emit() + def _scan_settings(self, data: bytes) -> None: + """Read SETTINGS entries as they stream; keep the last MAX_CONCURRENT_STREAMS.""" + self._scan -= len(data) + self._entry += data + whole = len(self._entry) - len(self._entry) % 6 + for at in range(0, whole, 6): + if self._entry[at : at + 2] == b"\x00\x03": + self._payload[:] = self._entry[at : at + 6] + del self._entry[:whole] + def _emit(self): payload = bytes(self._payload) self._payload.clear() self._after = self._on_frame(*self._frame, payload) or b"" -async def _server_preface(reader) -> bytes: - """The server's first frame, whole (RFC 9113 section 3.4: a SETTINGS frame).""" +async def _first_frame(reader) -> bytes: + """A peer's first frame, whole: its SETTINGS (RFC 9113 section 3.4), within the frame + size any peer may send before it has the other's.""" header = await reader.readexactly(_FRAME.size) high, low, *_ = _FRAME.unpack(header) - return header + await reader.readexactly(high << 16 | low) + length = high << 16 | low + if length > _INITIAL_MAX_FRAME: + raise _ProtocolError(f"a first frame of {length} bytes") + return header + await reader.readexactly(length) def _with_max_streams(frame: bytes, limit: int) -> bytes: @@ -600,14 +636,33 @@ async def idle(): return await asyncio.sleep(remaining if remaining > 0 else self.idle_timeout) - # The server's connection preface, a SETTINGS frame, before the client is read: so + # The client's SETTINGS, which completes its preface, goes to the server first: a + # server may wait for it before it sends its own. + opening_errors = ( + TimeoutError, + asyncio.IncompleteReadError, + ConnectionError, + _ProtocolError, + ) + try: + async with asyncio.timeout(self.read_timeout): + opening = await _first_frame(reader) + except opening_errors as exc: + _log.info("connect face: the client sent no SETTINGS: %r", exc) + upstream_writer.close() + return + client.feed(opening) + upstream_writer.write(opening) + # Then the server's preface, its SETTINGS, before the client is read further: so # the face knows the server's limit before the client's first call, and a client # of a server that advertises none learns the face's. try: async with asyncio.timeout(self.read_timeout): - preface = await _server_preface(upstream_reader) - except (TimeoutError, asyncio.IncompleteReadError, ConnectionError) as exc: - _log.info("connect face: the gRPC server sent no SETTINGS: %r", exc) + preface = await _first_frame(upstream_reader) + except opening_errors as exc: + reason = f"the gRPC server sent no SETTINGS within {self.read_timeout:g} s" + _log.warning("connect face: %s: %r", reason, exc) + writer.write(_goaway(0, _INTERNAL_ERROR, reason)) upstream_writer.close() return server.feed(preface) diff --git a/testdata/connect/test_connect_gen.py b/testdata/connect/test_connect_gen.py index 54ce722..d4151fe 100644 --- a/testdata/connect/test_connect_gen.py +++ b/testdata/connect/test_connect_gen.py @@ -478,6 +478,30 @@ async def test_a_refusal_the_service_passes_on_keeps_its_code_and_words(face): assert r.json() == {"code": "resource_exhausted", "message": PASSED_ON} +async def test_a_refusal_naming_the_faces_own_bound_is_still_the_services(): + # The face's bounds both 4 MiB, the same bound the service's client names: the + # face's channel refuses only past one byte more, so the number cannot be its own. + server = grpc.aio.server(interceptors=(Proof(),)) + server.add_generic_rpc_handlers((_handler(),)) + grpc_port = server.add_insecure_port("127.0.0.1:0") + await server.start() + listener = await connect_gen.serve( + 0, grpc_port, host="127.0.0.1", max_response_bytes=4194304, max_json_response_bytes=4194304 + ) + try: + async with httpx.AsyncClient() as client: + r = await client.post( + _url(listener, "Say"), + json={"text": "passing-on"}, + headers={"content-type": "application/json", **STAMP}, + ) + assert r.status_code == 429 + assert r.json() == {"code": "resource_exhausted", "message": PASSED_ON} + finally: + await listener.stop() + await server.stop(None) + + @pytest.mark.parametrize("option", ["max_response_bytes", "max_json_response_bytes"]) @pytest.mark.parametrize("value", [0, -1]) async def test_an_answer_bound_is_a_positive_size(option, value): @@ -609,7 +633,7 @@ async def drip(reader, writer): ) try: reader, writer = await asyncio.open_connection("127.0.0.1", listener.port) - writer.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n") + writer.write(PREFACE + SETTINGS) await writer.drain() got = read_frames(await asyncio.wait_for(reader.read(), 3)) assert [f[3] for f in got if f[0] == 0x6] == [bytes([n]) * 8 for n in range(6)] @@ -715,11 +739,11 @@ async def silent_upstream(): ``state["mid_frame"]`` (an event) makes it send half a frame first and the rest when the event is set.""" held, faces = [], [] - state = {"close_after": None, "received": bytearray(), "mid_frame": None} + state = {"close_after": None, "received": bytearray(), "mid_frame": None, "preface": SETTINGS} async def accept(reader, writer): held.append(writer) - writer.write(SETTINGS) # a server's connection preface + writer.write(state["preface"]) # a server's connection preface if state["mid_frame"] is not None: # Half a PING, the rest once the test sets the event: the server is mid-frame. ping = h2_frame(0x6, 0, 0, b"12345678") @@ -892,6 +916,54 @@ async def accept(reader, writer): upstream.close() +async def test_a_server_that_waits_for_the_clients_settings_is_not_stalled(): + # A server may send its SETTINGS only once it has the client's: the face forwards + # the client's first, so the two prefaces cross. + async def accept(reader, writer): + await reader.readexactly(len(PREFACE) + len(SETTINGS)) + writer.write(h2_frame(0x4, 0, 0, struct.pack(">HI", 0x3, 7))) + await writer.drain() + while await reader.read(65536): + pass + + upstream = await asyncio.start_server(accept, "127.0.0.1", 0) + listener = await connect_gen.serve( + 0, upstream.sockets[0].getsockname()[1], host="127.0.0.1", read_timeout=5 + ) + try: + _, got = await exchange(listener, PREFACE + SETTINGS, 1) + assert [_settings(f) for f in read_frames(got) if f[0] == 0x4] == [[(0x3, 7)]] + finally: + await listener.stop() + upstream.close() + + +async def test_a_server_that_sends_no_settings_gets_the_client_a_goaway(silent_upstream): + face, state = silent_upstream + state["preface"] = b"" # a server that never sends its SETTINGS + closed, got = await exchange(await face(read_timeout=0.5), PREFACE + SETTINGS, 2) + assert closed + ((kind, _, _, payload),) = read_frames(got) + last, code = struct.unpack_from(">II", payload) + assert (kind, last, code) == (0x7, 0, 0x2) + assert payload[8:] == b"the gRPC server sent no SETTINGS within 0.5 s" + + +async def test_the_servers_limit_is_found_anywhere_in_its_settings(): + # Past the first 256 bytes, and in a frame split across reads. + entries = b"".join(struct.pack(">HI", 0xF000 + n, n) for n in range(60)) + server_settings = h2_frame(0x4, 0, 0, entries + struct.pack(">HI", 0x3, 7)) + to_client, _ = await _first_frames(server_settings, PREFACE + SETTINGS) + assert _settings(to_client[0])[-1] == (0x3, 7) + assert (0x3, connect_gen.DEFAULT_MAX_OPEN_CALLS) not in _settings(to_client[0]) + + seen = [] + frames = connect_gen._Frames(lambda kind, flags, stream, payload: seen.append(payload)) + for at in range(len(server_settings)): + frames.feed(server_settings[at : at + 1]) + assert seen == [struct.pack(">HI", 0x3, 7)] and frames.at_boundary + + async def test_a_server_advertising_no_limit_is_given_the_faces(): # grpcio advertises no SETTINGS_MAX_CONCURRENT_STREAMS: the client is told the face's # DEFAULT_MAX_OPEN_CALLS in the server's own SETTINGS, so it queues within it. From 4ecf54339eef389e66e752f91eb5defa3ae81715 Mon Sep 17 00:00:00 2001 From: antoinetoussaint-byte <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Sun, 25 Oct 2026 22:56:25 -0400 Subject: [PATCH 2/2] fix(connect): every opening refusal is a valid server preface that names what happened Review r121 of #118: - B1: the GOAWAY for a server that sent no SETTINGS was the face's first frame to the client; RFC 9113 3.4 makes a server's first frame SETTINGS, and grpcio discarded the reason ("Expected SETTINGS frame as the first frame"). Every refusal of an opening is now an empty SETTINGS, then the GOAWAY. A real grpcio client logs "Got goaway [2] ... Debug Text: the gRPC server sent no SETTINGS within 0.5 s". - m1: the channel's receive bound is clamped to 2**31 - 1 (gRPC's channel arguments are C ints), so a bound at gRPC's largest starts again. - m2: the reason says what happened: no SETTINGS within N s, closed before its SETTINGS, a first frame of N bytes past the 16384 allowed, or a first frame that is not SETTINGS. - m3: the 16 KiB first-frame cap is tested in both directions; a client's oversized first frame gets SETTINGS + GOAWAY(FRAME_SIZE_ERROR). - m4: only the server's direction scans SETTINGS; the client's skips them as before (scan_bench: 14 GB/s, main 15 GB/s). - m5: a first frame that is not a SETTINGS frame on stream 0 without ACK is refused, from either peer, with GOAWAY(PROTOCOL_ERROR) naming it; the client's never reaches the server. Tests (each fails on 43edffd): test_a_server_that_sends_no_settings_gets_the_client_a_goaway (silent, closed, oversized, not-settings), test_a_grpc_client_hears_why_the_server_never_answered (a grpcio client in its own process), test_a_client_that_sends_no_settings_is_told_why, test_a_client_whose_first_frame_is_not_its_settings_is_refused (oversized, headers, settings-ack), test_a_bound_at_grpcs_largest_starts, and the client direction in test_the_servers_limit_is_found_anywhere_in_its_settings. Co-Authored-By: Claude Opus 5.5 --- templates/connect/connect_gen.py | 70 +++++++++--- testdata/connect/test_connect_gen.py | 155 +++++++++++++++++++++++++-- 2 files changed, 198 insertions(+), 27 deletions(-) diff --git a/templates/connect/connect_gen.py b/templates/connect/connect_gen.py index 27b9f89..1d6e230 100644 --- a/templates/connect/connect_gen.py +++ b/templates/connect/connect_gen.py @@ -96,9 +96,14 @@ _END_STREAM_FLAG, _ACK_FLAG, _END_HEADERS_FLAG = 0x1, 0x1, 0x4 _MAX_CONCURRENT_STREAMS = 0x3 _PUSH_PROMISE = 0x5 -_PROTOCOL_ERROR, _INTERNAL_ERROR, _CANCEL, _ENHANCE_YOUR_CALM = 0x1, 0x2, 0x8, 0xB -# The largest frame a peer may send before it has the other's SETTINGS (RFC 9113 4.2). +_PROTOCOL_ERROR, _INTERNAL_ERROR, _FRAME_SIZE_ERROR = 0x1, 0x2, 0x6 +_CANCEL, _ENHANCE_YOUR_CALM = 0x8, 0xB +# The largest frame a client may send before it has the server's SETTINGS (RFC 9113 +# 4.2); the face holds the server's first frame to it too, though the client's SETTINGS +# could allow more (a SETTINGS frame that size would carry over 2,700 entries). _INITIAL_MAX_FRAME = 16384 +# gRPC's channel arguments are C ints. +_GRPC_MAX_INT = 2**31 - 1 # Calls one relayed connection may hold open when the gRPC server advertises no # SETTINGS_MAX_CONCURRENT_STREAMS (grpcio advertises none unless the service sets # grpc.max_concurrent_streams, as the scaffold's server does). The face advertises it to @@ -294,11 +299,12 @@ class _AnswerBounds: them as they are, ``json`` for one that converts them, checked before converting. ``received`` is the channel's own receive bound: one byte past both, so the face checks every bound itself, and the number grpcio names when it refuses past it is - one no service's client has by chance with the same text.""" + one no service's client has by chance with the same text. A bound at gRPC's largest + (2**31 - 1, its "unlimited") keeps the channel there.""" def __init__(self, proto: int, json: int): self.proto, self.json = proto, json - self.received = max(proto, json) + 1 + self.received = min(max(proto, json) + 1, _GRPC_MAX_INT) def of(self, codec: _Codec) -> tuple[int, str]: if codec.name == "proto": @@ -344,13 +350,15 @@ class _Frames: where frames end in the data, each with the bytes ``on_frame`` returned, to be written right after that frame in this direction. The payload passed is, for a SETTINGS frame, its last SETTINGS_MAX_CONCURRENT_STREAMS entry if it has one (the frame is - scanned whole as it streams, holding one entry), for a GOAWAY its first + scanned whole as it streams, holding one entry, when ``scan_settings``: the server's + direction only, since only the server's limit matters), for a GOAWAY its first ``_CONTROL_PAYLOAD`` bytes, and empty otherwise. A ``_ProtocolError`` from ``on_frame`` stops the feed; it is kept in ``error`` and the ends before the offending frame are returned.""" - def __init__(self, on_frame): + def __init__(self, on_frame, *, scan_settings: bool = False): self._on_frame = on_frame + self._scan_settings_frames = scan_settings self._header = bytearray() self._frame = None # (type, flags, stream) while its payload is read self._after = b"" # what on_frame returned for it @@ -429,7 +437,7 @@ def _start(self, high, low, kind, flags, stream): if kind in (_HEADERS, _PUSH_PROMISE, _CONTINUATION): self._in_block = not flags & _END_HEADERS_FLAG self._keep = min(length, _CONTROL_PAYLOAD) if kind == _GOAWAY else 0 - self._scan = length if kind == _SETTINGS else 0 + self._scan = length if kind == _SETTINGS and self._scan_settings_frames else 0 self._skip = length - self._keep - self._scan if not (self._keep or self._scan): self._emit() @@ -450,17 +458,43 @@ def _emit(self): self._after = self._on_frame(*self._frame, payload) or b"" +class _FrameSizeError(_ProtocolError): + """A first frame past ``_INITIAL_MAX_FRAME``.""" + + async def _first_frame(reader) -> bytes: - """A peer's first frame, whole: its SETTINGS (RFC 9113 section 3.4), within the frame - size any peer may send before it has the other's.""" + """A peer's first frame, whole, which RFC 9113 section 3.4 makes its SETTINGS: a + SETTINGS frame on stream 0, not an acknowledgement, within ``_INITIAL_MAX_FRAME``.""" header = await reader.readexactly(_FRAME.size) - high, low, *_ = _FRAME.unpack(header) + high, low, kind, flags, stream = _FRAME.unpack(header) length = high << 16 | low if length > _INITIAL_MAX_FRAME: - raise _ProtocolError(f"a first frame of {length} bytes") + raise _FrameSizeError( + f"a first frame of {length} bytes, past the {_INITIAL_MAX_FRAME} " + "allowed before SETTINGS" + ) + if kind != _SETTINGS or flags & _ACK_FLAG or stream: + raise _ProtocolError( + f"a first frame of type {kind} (flags {flags}, stream {stream}), not SETTINGS" + ) return header + await reader.readexactly(length) +def _opening_refusal(exc: BaseException, peer: str, timeout: float) -> str: + """Why a peer's opening failed, in words an operator can act on.""" + if isinstance(exc, TimeoutError): + return f"the {peer} sent no SETTINGS within {timeout:g} s" + if isinstance(exc, _ProtocolError): + return f"the {peer} sent {exc}" + return f"the {peer} closed the connection before its SETTINGS" + + +def _refuse_opening(writer, code: int, reason: str) -> None: + """Tell the client why its connection ends, as a server's first frames must: SETTINGS + first (RFC 9113 section 3.4: a client discards anything else), then the GOAWAY.""" + writer.write(_h2_frame(_SETTINGS, 0, 0) + _goaway(0, code, reason)) + + def _with_max_streams(frame: bytes, limit: int) -> bytes: """A server SETTINGS frame that advertises no SETTINGS_MAX_CONCURRENT_STREAMS, with ``limit`` added: the client then holds its calls within what the face enforces. The @@ -560,7 +594,7 @@ def from_server(kind, flags, stream, payload): calls.difference_update({call for call in calls if call > last_stream}) return None - client, server = _Frames(from_client), _Frames(from_server) + client, server = _Frames(from_client), _Frames(from_server, scan_settings=True) def tell_client(frame): nonlocal told @@ -648,7 +682,11 @@ async def idle(): async with asyncio.timeout(self.read_timeout): opening = await _first_frame(reader) except opening_errors as exc: - _log.info("connect face: the client sent no SETTINGS: %r", exc) + reason = _opening_refusal(exc, "client", self.read_timeout) + _log.info("connect face: %s", reason) + if isinstance(exc, (TimeoutError, _ProtocolError)): + code = _FRAME_SIZE_ERROR if isinstance(exc, _FrameSizeError) else _PROTOCOL_ERROR + _refuse_opening(writer, code, reason) upstream_writer.close() return client.feed(opening) @@ -660,9 +698,9 @@ async def idle(): async with asyncio.timeout(self.read_timeout): preface = await _first_frame(upstream_reader) except opening_errors as exc: - reason = f"the gRPC server sent no SETTINGS within {self.read_timeout:g} s" - _log.warning("connect face: %s: %r", reason, exc) - writer.write(_goaway(0, _INTERNAL_ERROR, reason)) + reason = _opening_refusal(exc, "gRPC server", self.read_timeout) + _log.warning("connect face: %s", reason) + _refuse_opening(writer, _INTERNAL_ERROR, reason) upstream_writer.close() return server.feed(preface) diff --git a/testdata/connect/test_connect_gen.py b/testdata/connect/test_connect_gen.py index d4151fe..2aed1f6 100644 --- a/testdata/connect/test_connect_gen.py +++ b/testdata/connect/test_connect_gen.py @@ -4,6 +4,8 @@ import asyncio import base64 import collections +import os +import sys import json import struct @@ -595,8 +597,9 @@ async def test_connect_json_ignores_unknown_fields(face): assert r.json() == {"text": "hi|maya", "count": 1} -async def test_a_silent_grpc_connection_is_closed_after_the_idle_bound(): - # An upstream that accepts and never answers: only the face's own bound ends it. +async def test_a_client_that_sends_no_settings_is_told_why(): + # A client that sends the preface's magic and then nothing: the face's read bound + # ends it, and the client is told why, after the SETTINGS a server's first frame is. upstream = await asyncio.start_server(lambda r, w: None, "127.0.0.1", 0) listener = await connect_gen.serve( 0, @@ -609,7 +612,10 @@ async def test_a_silent_grpc_connection_is_closed_after_the_idle_bound(): reader, writer = await asyncio.open_connection("127.0.0.1", listener.port) writer.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n") await writer.drain() - assert await asyncio.wait_for(reader.read(), 3) == b"" + sent = read_frames(await asyncio.wait_for(reader.read(), 3)) + assert [f[:3] for f in sent] == [(0x4, 0, 0), (0x7, 0, 0)] + reason = b"the client sent no SETTINGS within 0.5 s" + assert sent[1][3] == struct.pack(">II", 0, 0x1) + reason writer.close() finally: await listener.stop() @@ -938,15 +944,134 @@ async def accept(reader, writer): upstream.close() -async def test_a_server_that_sends_no_settings_gets_the_client_a_goaway(silent_upstream): +def _refusal(got): + """The face's refusal of a connection: (GOAWAY code, reason), after an empty SETTINGS.""" + frames = read_frames(got) + assert [f[:3] for f in frames] == [(0x4, 0, 0), (0x7, 0, 0)], frames + assert frames[0][3] == b"" + last, code = struct.unpack_from(">II", frames[1][3]) + assert last == 0 + return code, frames[1][3][8:].decode() + + +@pytest.mark.parametrize( + "preface, reason", + [ + (None, "the gRPC server sent no SETTINGS within 0.5 s"), + (b"", "the gRPC server closed the connection before its SETTINGS"), + ( + h2_frame(0x4, 0, 0, b"\0" * 16386), + "the gRPC server sent a first frame of 16386 bytes, past the 16384 allowed " + "before SETTINGS", + ), + ( + h2_frame(0x6, 0, 0, b"\0" * 8), + "the gRPC server sent a first frame of type 6 (flags 0, stream 0), not SETTINGS", + ), + ], + ids=["silent", "closed", "oversized", "not-settings"], +) +async def test_a_server_that_sends_no_settings_gets_the_client_a_goaway(preface, reason): + # Whatever went wrong, the client hears why, after an empty SETTINGS (RFC 9113 3.4: + # a server's first frame; a client discards anything else). + async def accept(reader, writer): + if preface is not None: + writer.write(preface) + await writer.drain() + if preface == b"": + writer.close() + return + while await reader.read(65536): + pass + + upstream = await asyncio.start_server(accept, "127.0.0.1", 0) + listener = await connect_gen.serve( + 0, upstream.sockets[0].getsockname()[1], host="127.0.0.1", read_timeout=0.5 + ) + try: + closed, got = await exchange(listener, PREFACE + SETTINGS, 2) + assert closed + assert _refusal(got) == (0x2, reason) + finally: + await listener.stop() + upstream.close() + + +_GRPC_CLIENT = """ +import sys, grpc +channel = grpc.insecure_channel(sys.argv[1]) +try: + channel.unary_unary("/echo.v1.Echo/Say")(b"", timeout=2) +except grpc.RpcError as error: + print(error.code()) +""" + + +async def test_a_grpc_client_hears_why_the_server_never_answered(): + # A real gRPC client, in its own process so its transport log can be read: it parses + # the face's refusal (a GOAWAY first would be "Expected SETTINGS frame") and logs + # its reason. + async def accept(reader, writer): + while await reader.read(65536): + pass + + upstream = await asyncio.start_server(accept, "127.0.0.1", 0) + listener = await connect_gen.serve( + 0, upstream.sockets[0].getsockname()[1], host="127.0.0.1", read_timeout=0.5 + ) + try: + client = await asyncio.create_subprocess_exec( + sys.executable, + "-c", + _GRPC_CLIENT, + f"127.0.0.1:{listener.port}", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env={**os.environ, "GRPC_VERBOSITY": "info"}, + ) + out, err = await asyncio.wait_for(client.communicate(), 30) + log = err.decode() + assert "Expected SETTINGS frame" not in log + assert "Debug Text: the gRPC server sent no SETTINGS within 0.5 s" in log, log[-2000:] + finally: + await listener.stop() + upstream.close() + + +@pytest.mark.parametrize( + "opening, code, reason", + [ + ( + h2_frame(0x4, 0, 0, b"\0" * 16386), + 0x6, + "the client sent a first frame of 16386 bytes, past the 16384 allowed before SETTINGS", + ), + ( + h2_frame(0x1, 0x4, 1, b"\x82"), + 0x1, + "the client sent a first frame of type 1 (flags 4, stream 1), not SETTINGS", + ), + ( + h2_frame(0x4, 0x1, 0), + 0x1, + "the client sent a first frame of type 4 (flags 1, stream 0), not SETTINGS", + ), + ], + ids=["oversized", "headers", "settings-ack"], +) +async def test_a_client_whose_first_frame_is_not_its_settings_is_refused( + silent_upstream, opening, code, reason +): face, state = silent_upstream - state["preface"] = b"" # a server that never sends its SETTINGS - closed, got = await exchange(await face(read_timeout=0.5), PREFACE + SETTINGS, 2) + closed, got = await exchange(await face(), PREFACE + opening, 2) assert closed - ((kind, _, _, payload),) = read_frames(got) - last, code = struct.unpack_from(">II", payload) - assert (kind, last, code) == (0x7, 0, 0x2) - assert payload[8:] == b"the gRPC server sent no SETTINGS within 0.5 s" + assert _refusal(got) == (code, reason) + assert bytes(state["received"]) == PREFACE # the frame never reached the server + + +async def test_a_bound_at_grpcs_largest_starts(): + listener = await connect_gen.serve(0, 1, host="127.0.0.1", max_response_bytes=2**31 - 1) + await listener.stop() async def test_the_servers_limit_is_found_anywhere_in_its_settings(): @@ -958,11 +1083,19 @@ async def test_the_servers_limit_is_found_anywhere_in_its_settings(): assert (0x3, connect_gen.DEFAULT_MAX_OPEN_CALLS) not in _settings(to_client[0]) seen = [] - frames = connect_gen._Frames(lambda kind, flags, stream, payload: seen.append(payload)) + frames = connect_gen._Frames( + lambda kind, flags, stream, payload: seen.append(payload), scan_settings=True + ) for at in range(len(server_settings)): frames.feed(server_settings[at : at + 1]) assert seen == [struct.pack(">HI", 0x3, 7)] and frames.at_boundary + # The client's direction only skips them: its SETTINGS cost nothing to read. + seen.clear() + frames = connect_gen._Frames(lambda kind, flags, stream, payload: seen.append(payload)) + frames.feed(server_settings) + assert seen == [b""] and frames.at_boundary + async def test_a_server_advertising_no_limit_is_given_the_faces(): # grpcio advertises no SETTINGS_MAX_CONCURRENT_STREAMS: the client is told the face's