diff --git a/templates/connect/connect_gen.py b/templates/connect/connect_gen.py index 364549d..1d6e230 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,14 @@ _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, _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 @@ -113,7 +123,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 +296,15 @@ 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. 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) + self.received = min(max(proto, json) + 1, _GRPC_MAX_INT) def of(self, codec: _Codec) -> tuple[int, str]: if codec.name == "proto": @@ -333,19 +348,25 @@ 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, 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 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 +374,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 +400,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 +419,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 +436,63 @@ 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 and self._scan_settings_frames 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).""" +class _FrameSizeError(_ProtocolError): + """A first frame past ``_INITIAL_MAX_FRAME``.""" + + +async def _first_frame(reader) -> bytes: + """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) - return header + await reader.readexactly(high << 16 | low) + high, low, kind, flags, stream = _FRAME.unpack(header) + length = high << 16 | low + if length > _INITIAL_MAX_FRAME: + 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: @@ -524,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 @@ -600,14 +670,37 @@ 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: + 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) + 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 = _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 54ce722..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 @@ -478,6 +480,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): @@ -571,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, @@ -585,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() @@ -609,7 +639,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 +745,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 +922,181 @@ 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() + + +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 + closed, got = await exchange(await face(), PREFACE + opening, 2) + assert closed + 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(): + # 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), 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 # DEFAULT_MAX_OPEN_CALLS in the server's own SETTINGS, so it queues within it.