diff --git a/templates/connect/connect_gen.py b/templates/connect/connect_gen.py index 4428566..1c86cc4 100644 --- a/templates/connect/connect_gen.py +++ b/templates/connect/connect_gen.py @@ -18,11 +18,35 @@ end to end, so they are served on the gRPC protocol only. Request bodies are bounded by ``max_message_bytes`` (gRPC's 4 MiB unless the service -raises it) before they are read, and every read is bounded in time: an HTTP/1.1 read by -``read_timeout`` (``idle_timeout`` between requests), and a relayed gRPC connection is -closed once it has had no open call for ``idle_timeout``, as Go's HTTP/2 server bounds an -idle connection. A call in flight is never cut by the face, however long its handler -takes to answer: its deadline and the gRPC server's own policy bound it. +raises it) before they are read. Answers are bounded apart, on the proto the service +returns, by the codec the caller asked for: + +- ``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. +- ``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. + +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, +Connect and gRPC-Web return what the gRPC server returns, and a refusal the service sends +keeps its code and words. Both bounds are positive numbers of bytes, keyword options to +``serve`` and ``start``; a relayed gRPC call keeps the service's and the caller's own +bounds. The face does not bound how many calls are in flight, as the gRPC server does +not: each holds its answer until its client reads it. + +Every read is bounded in time: an HTTP/1.1 read by ``read_timeout`` (``idle_timeout`` +between requests), and a relayed gRPC connection is closed once it has had no open call +for ``idle_timeout``, as Go's HTTP/2 server bounds an idle connection. A call in flight +is never cut by the face, however long its handler takes to answer: its deadline and the +gRPC server's own policy bound it. A JSON request field the service's proto does not know is ignored, as connect-go's codec ignores it, so a client built from a newer proto calls a Python service and a Go one alike. @@ -57,6 +81,8 @@ # The port a deployment hands the face (CODEFLY_CONNECT_PORT); ``start`` reads only the variable. DEFAULT_PORT = 8081 DEFAULT_MAX_MESSAGE_BYTES = 4 * 1024 * 1024 +DEFAULT_MAX_RESPONSE_BYTES = 16 * 1024 * 1024 +DEFAULT_MAX_JSON_RESPONSE_BYTES = 4 * 1024 * 1024 DEFAULT_READ_TIMEOUT_S = 30.0 DEFAULT_IDLE_TIMEOUT_S = 120.0 @@ -111,6 +137,10 @@ ) _TRANSPORT_PREFIXES = ("connect-", "grpc-", "x-grpc-web", ":") _GRPC_TIMEOUT = re.compile(r"^(\d{1,8})([HMSmun])$") +# grpcio's refusal of an answer past the face's own receive bound (its channel is the client). +_ANSWER_PAST_BOUND = re.compile( + r"Stream removed \(CLIENT: Received message larger than max \((\d+) vs\. (\d+)\)\)" +) _GRPC_UNITS = {"H": 3600.0, "M": 60.0, "S": 1.0, "m": 1e-3, "u": 1e-6, "n": 1e-9} _CODES = { grpc.StatusCode.CANCELLED: ("canceled", 499), @@ -183,22 +213,43 @@ async def serve( host: str = "0.0.0.0", grpc_host: str = "127.0.0.1", max_message_bytes: int = DEFAULT_MAX_MESSAGE_BYTES, + max_response_bytes: int | None = None, + max_json_response_bytes: int | None = None, read_timeout: float = DEFAULT_READ_TIMEOUT_S, idle_timeout: float = DEFAULT_IDLE_TIMEOUT_S, ) -> ConnectFace: """Start the Connect face on ``host``:``port`` (0 for an ephemeral one) over the grpc.aio server listening on ``grpc_host``:``grpc_port``, and return it started. Call it on the loop that serves the gRPC server, once that server listens.""" + bounds = _AnswerBounds( + _positive( + "max_response_bytes", max_response_bytes, DEFAULT_MAX_RESPONSE_BYTES, max_message_bytes + ), + _positive( + "max_json_response_bytes", + max_json_response_bytes, + DEFAULT_MAX_JSON_RESPONSE_BYTES, + max_message_bytes, + ), + ) target = f"{grpc_host}:{grpc_port}" channel = grpc.aio.insecure_channel( target, options=[ - ("grpc.max_receive_message_length", max_message_bytes), + # The face receives the service's answers and sends it the requests. The + # channel stops an answer past both bounds before it is buffered; the face + # checks the bound of the call's codec itself. + ("grpc.max_receive_message_length", bounds.received), ("grpc.max_send_message_length", max_message_bytes), ], ) connection = _Connection( - channel, (grpc_host, grpc_port), max_message_bytes, read_timeout, idle_timeout + channel, + (grpc_host, grpc_port), + max_message_bytes, + bounds, + read_timeout, + idle_timeout, ) face: ConnectFace @@ -221,6 +272,29 @@ async def accepted(reader, writer): return face +def _positive(name: str, value: int | None, default: int, message_bound: int) -> int: + """A bound as given, or by default the larger of ``default`` and the message bound.""" + if value is None: + return max(default, message_bound) + if value <= 0: + raise ValueError(f"{name} must be a positive number of bytes, not {value}") + return value + + +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.""" + + def __init__(self, proto: int, json: int): + self.proto, self.json = proto, json + self.received = max(proto, json) + + def of(self, codec: _Codec) -> tuple[int, str]: + if codec.name == "proto": + return self.proto, "max_response_bytes" + return self.json, "max_json_response_bytes" + + async def start(grpc_port: int, **options) -> ConnectFace | None: """Start the face the agent's generated ``src/main.py`` starts: on ``CODEFLY_CONNECT_PORT`` over the gRPC server on ``grpc_port``, with ``serve``'s @@ -343,8 +417,9 @@ def _emit(self): class _Connection: - def __init__(self, channel, grpc_address, limit, read_timeout, idle_timeout): + def __init__(self, channel, grpc_address, limit, answer_bounds, read_timeout, idle_timeout): self.channel, self.grpc_address, self.limit = channel, grpc_address, limit + self.answer_bounds = answer_bounds self.read_timeout, self.idle_timeout = read_timeout, idle_timeout async def serve(self, reader, writer): @@ -598,7 +673,16 @@ async def _dispatch(self, exchange, request, body): codec = protocol.codec if codec is None: raise _Plain(415, f"unsupported codec in {content_type!r}") - call = _Call(self.channel, path, method, codec, request.headers, headers, self.limit) + call = _Call( + self.channel, + path, + method, + codec, + request.headers, + headers, + self.limit, + self.answer_bounds, + ) await protocol.answer(exchange, call, body, headers) @@ -643,9 +727,11 @@ class _Call: """One call dispatched through the gRPC server: request bytes in the protocol's codec, answers out in it.""" - def __init__(self, channel, path, method, codec, raw_headers, headers, limit): + def __init__(self, channel, path, method, codec, raw_headers, headers, limit, answer_bounds): self.channel, self.path, self.method, self.codec = channel, path, method, codec self.limit = limit + self.answer_bound, self.answer_bound_name = answer_bounds.of(codec) + self.received_bound = answer_bounds.received self.input = message_factory.GetMessageClass(method.input_type) self.output = message_factory.GetMessageClass(method.output_type) self.metadata = _metadata(raw_headers) @@ -678,6 +764,32 @@ def decompress(self, data: bytes, encoding: str | None) -> bytes: ) return payload + def answer(self, data: bytes) -> bytes: + """An answer in the call's codec, within the bound of that codec, checked before it + is converted: the conversion is what costs (see the module's docstring).""" + if len(data) > self.answer_bound: + raise self._too_large(len(data)) + return self.codec.response(self.output, data) + + def failed(self, error: grpc.aio.AioRpcError) -> CallError: + """The call's status. An answer past the channel's own receive bound (one past + every bound of the face) is refused by grpcio before it is buffered, with only its + text to say so: it is the face's refusal when that text is grpcio's own and names + exactly the bound the face set. A service passing on a refusal from its own client + names that client's bound, and keeps its code and words.""" + past = _ANSWER_PAST_BOUND.fullmatch(error.details() or "") + own = past is not None and int(past.group(2)) == self.received_bound + if error.code() is grpc.StatusCode.RESOURCE_EXHAUSTED and own: + return self._too_large(int(past.group(1))) + return CallError.of(error) + + def _too_large(self, size: int) -> CallError: + return CallError( + grpc.StatusCode.FAILED_PRECONDITION, + f"the answer ({size} bytes) exceeds the Connect face's {self.answer_bound_name} " + f"({self.answer_bound})", + ) + async def unary(self, message: bytes): """The answer and the header and trailer metadata of a unary call.""" request = self.codec.request(self.input, message) @@ -687,12 +799,12 @@ async def unary(self, message: bytes): try: answer = await call return ( - self.codec.response(self.output, answer), + self.answer(answer), tuple(await call.initial_metadata() or ()), tuple(await call.trailing_metadata() or ()), ) except grpc.aio.AioRpcError as exc: - raise CallError.of(exc) from None + raise self.failed(exc) from None def stream(self, message: bytes): request = self.codec.request(self.input, message) @@ -852,11 +964,13 @@ async def _relay(exchange, call, stream): return stream.trailer try: async for answer in stream: - data = call.codec.response(call.output, answer) + data = call.answer(answer) await exchange.send(h11.Data(data=_ENVELOPE.pack(0, len(data)) + data)) return tuple(await stream.trailing_metadata() or ()) except grpc.aio.AioRpcError as exc: - return CallError.of(exc) + return call.failed(exc) + except CallError as error: + return error finally: stream.cancel() diff --git a/testdata/connect/test_connect_gen.py b/testdata/connect/test_connect_gen.py index 83e9462..cdf2e89 100644 --- a/testdata/connect/test_connect_gen.py +++ b/testdata/connect/test_connect_gen.py @@ -57,6 +57,11 @@ async def refuse(request, context): return await continuation(details) +# What a service passes on from its own gRPC client when a downstream answer is past +# that client's bound: grpcio's own words, naming that client's bound. +PASSED_ON = "Stream removed (CLIENT: Received message larger than max (5000005 vs. 4194304))" + + async def say(request, context): metadata = dict(context.invocation_metadata()) if request.text == "deny": @@ -64,12 +69,20 @@ async def say(request, context): await context.abort(grpc.StatusCode.PERMISSION_DENIED, "denied: not yours") if request.text == "slow": await asyncio.sleep(2) + if request.text == "big": + # An answer of ``count`` bytes from a request of a few. + return Msg(text="x" * request.count) + if request.text == "passing-on": + await context.abort(grpc.StatusCode.RESOURCE_EXHAUSTED, PASSED_ON) await context.send_initial_metadata((("x-answer", "header"),)) context.set_trailing_metadata((("x-trail", "trailer"), ("x-raw-bin", b"\x00\xff"))) return Msg(text=f"{request.text}|{metadata.get('x-caller', '')}", count=request.count) async def count(request, context): + if request.text == "big": + yield Msg(text="x" * request.count) + return for n in range(request.count): yield Msg(text=request.text, count=n) if request.text == "fail": @@ -268,6 +281,264 @@ async def test_http_refusals(face): ) assert (unknown.status_code, get.status_code, media.status_code) == (404, 405, 415) assert big.status_code == 413 + assert big.text == "content length (4096) exceeds the message bound" + + +async def test_an_oversized_request_never_reaches_the_service(face): + # A separate response bound leaves requests past 1024 bytes refused. + async with httpx.AsyncClient() as client: + r = await client.post( + _url(face, "Say"), + content=(request := Msg(text="x" * 2048).SerializeToString()), + headers={"content-type": "application/proto", **STAMP}, + ) + assert r.status_code == 413 + assert r.text == f"content length ({len(request)}) exceeds the message bound" + + +@pytest.mark.parametrize("size", [393 * 1024, 5 * 1024 * 1024]) +async def test_an_answer_larger_than_the_request_bound_is_answered(face, size): + # The face bounds requests at 1024 bytes; embedding-sized answers and answers + # above grpcio's default 4 MiB receive bound still come back as proto, and JSON within + # max_json_response_bytes (4 MiB by default: the conversion is what costs). + big = Msg(text="big", count=size) + async with httpx.AsyncClient() as client: + unary = await client.post( + _url(face, "Say"), + content=big.SerializeToString(), + headers={"content-type": "application/proto", **STAMP}, + ) + as_json = await client.post( + _url(face, "Say"), + json={"text": "big", "count": size}, + headers={"content-type": "application/json", **STAMP}, + ) + web = await client.post( + _url(face, "Say"), + content=envelope(big.SerializeToString()), + headers={"content-type": "application/grpc-web+proto", **STAMP}, + ) + stream = await client.post( + _url(face, "Count"), + content=envelope(big.SerializeToString()), + headers={"content-type": "application/connect+proto", **STAMP}, + ) + assert unary.status_code == 200, unary.text + assert Msg.FromString(unary.content).text == "x" * size + if size <= connect_gen.DEFAULT_MAX_JSON_RESPONSE_BYTES: + assert as_json.status_code == 200, as_json.text + assert as_json.json() == {"text": "x" * size} + else: + assert as_json.status_code == 400 + assert "max_json_response_bytes (4194304)" in as_json.json()["message"] + out = frames(web.content) + assert Msg.FromString(out[0][1]).text == "x" * size + assert "grpc-status: 0" in out[-1][1].decode() + out = frames(stream.content) + assert Msg.FromString(out[0][1]).text == "x" * size + assert out[-1] == (0x02, b"{}") + + +@pytest.mark.parametrize( + "media", + [ + "application/json", + "application/proto", + "application/connect+proto", + "application/connect+json", + "application/grpc-web+proto", + ], +) +async def test_max_response_bytes_bounds_answers_when_set(media): + server = grpc.aio.server() + server.add_generic_rpc_handlers((_handler(),)) + grpc_port = server.add_insecure_port("127.0.0.1:0") + await server.start() + listener = None + try: + listener = await connect_gen.serve( + 0, + grpc_port, + host="127.0.0.1", + max_message_bytes=1024, + max_response_bytes=2048, + max_json_response_bytes=2048, + ) + bound = "max_json_response_bytes" if "json" in media else "max_response_bytes" + streaming = media.startswith("application/connect+") + enveloped = streaming or "grpc-web" in media + async with httpx.AsyncClient() as client: + for size in (1024, 4096): + message = Msg(text="big", count=size) + body = ( + json.dumps({"text": "big", "count": size}).encode() + if "json" in media + else message.SerializeToString() + ) + r = await client.post( + _url(listener, "Count" if streaming else "Say"), + content=envelope(body) if enveloped else body, + headers={"content-type": media}, + ) + if size == 1024: + assert r.status_code == 200, r.text + data = frames(r.content)[0][1] if enveloped else r.content + text = ( + json.loads(data)["text"] if "json" in media else Msg.FromString(data).text + ) + assert text == "x" * size + elif "grpc-web" in media: + trailers = frames(r.content)[-1][1].decode() + assert "grpc-status: 9" in trailers # FAILED_PRECONDITION + assert "max_response_bytes (2048)" in trailers + else: + if streaming: + assert r.status_code == 200 + assert len(frames(r.content)) == 1 + error = json.loads(frames(r.content)[0][1])["error"] + else: + # Not 429: the same call is refused the same way when retried. + assert r.status_code == 400 + error = r.json() + assert error["code"] == "failed_precondition" + assert error["message"] == ( + f"the answer (4099 bytes) exceeds the Connect face's {bound} (2048)" + ), error + finally: + if listener is not None: + await listener.stop() + await server.stop(None) + + +@pytest.mark.parametrize("media", ["application/json", "application/connect+json"]) +async def test_the_json_bound_is_checked_before_converting(media, monkeypatch): + # The JSON bound applies to the proto the service returns, before the conversion that + # would cost many times it: an answer past it is never converted, and the same + # answer as proto, within max_response_bytes, still comes back. + server = grpc.aio.server() + server.add_generic_rpc_handlers((_handler(),)) + grpc_port = server.add_insecure_port("127.0.0.1:0") + await server.start() + converted = [] + to_dict = connect_gen.json_format.MessageToDict + monkeypatch.setattr( + connect_gen.json_format, + "MessageToDict", + lambda message, **kw: converted.append(message.ByteSize()) or to_dict(message, **kw), + ) + listener = None + try: + listener = await connect_gen.serve( + 0, + grpc_port, + host="127.0.0.1", + max_message_bytes=1024, + max_response_bytes=2**20, + max_json_response_bytes=2048, + ) + streaming = media.startswith("application/connect+") + body = json.dumps({"text": "big", "count": 4096}).encode() + async with httpx.AsyncClient() as client: + proto = await client.post( + _url(listener, "Say"), + content=Msg(text="big", count=4096).SerializeToString(), + headers={"content-type": "application/proto"}, + ) + r = await client.post( + _url(listener, "Count" if streaming else "Say"), + content=envelope(body) if streaming else body, + headers={"content-type": media}, + ) + assert proto.status_code == 200 and len(proto.content) == 4099 + error = json.loads(frames(r.content)[-1][1])["error"] if streaming else r.json() + assert r.status_code == (200 if streaming else 400) + assert error == { + "code": "failed_precondition", + "message": ( + "the answer (4099 bytes) exceeds the Connect face's max_json_response_bytes (2048)" + ), + } + assert 4099 not in converted + finally: + if listener is not None: + await listener.stop() + await server.stop(None) + + +async def test_a_refusal_the_service_passes_on_keeps_its_code_and_words(face): + # A service passing on its own client's receive-limit refusal (grpcio's words, its + # client's bound) is not the face refusing an answer: code and words stay the server's. + async with httpx.AsyncClient() as client: + r = await client.post( + _url(face, "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} + + +@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): + listener = None + try: + with pytest.raises(ValueError, match=option): + listener = await connect_gen.serve(0, 1, host="127.0.0.1", **{option: value}) + finally: + if listener is not None: + await listener.stop() + + +async def test_a_raised_message_bound_raises_the_answer_bound(): + # max_response_bytes defaults to the larger of DEFAULT_MAX_RESPONSE_BYTES and + # max_message_bytes: a service that raised its bound to 100 MiB answers 70 MiB. + server = grpc.aio.server(options=[("grpc.max_send_message_length", 100 * 2**20)]) + 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_message_bytes=100 * 2**20 + ) + try: + size = 70 * 2**20 + async with httpx.AsyncClient(timeout=60) as client: + r = await client.post( + _url(listener, "Say"), + content=Msg(text="big", count=size).SerializeToString(), + headers={"content-type": "application/proto"}, + ) + assert r.status_code == 200, r.text[:200] + assert len(Msg.FromString(r.content).text) == size + finally: + await listener.stop() + await server.stop(None) + + +async def test_the_default_answer_bound_is_16_mib(): + server = grpc.aio.server() + 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") + try: + async with httpx.AsyncClient(timeout=60) as client: + within = await client.post( + _url(listener, "Say"), + content=Msg(text="big", count=16 * 2**20 - 8).SerializeToString(), + headers={"content-type": "application/proto"}, + ) + past = await client.post( + _url(listener, "Say"), + content=Msg(text="big", count=16 * 2**20).SerializeToString(), + headers={"content-type": "application/proto"}, + ) + assert within.status_code == 200 + assert past.status_code == 400 + assert "max_response_bytes (16777216)" in past.json()["message"] + finally: + await listener.stop() + await server.stop(None) async def test_one_connection_serves_several_calls(face):