fix: limit request body size (#542)

* fix: limit request body size

* chore: format

* format

* fix test

* fix test

Author: polvalente

grpc/test/grpc/integration/server_test.exs

@@ -557,6 +557,8 @@ defmodule GRPC.Integration.ServerTest do
 
         {:ok, conn_pid} = :gun.open(~c"localhost", port)
 
+        assert_receive {:gun_up, ^conn_pid, :http}
+
         stream_ref =
           :gun.get(conn_pid, "/v1/messages/#{name}", [
             {"accept", "application/json"}

grpc_server/lib/grpc/server.ex

@@ -434,7 +434,14 @@ defmodule GRPC.Server do
   @spec start_endpoint(atom(), non_neg_integer(), Keyword.t()) ::
           {atom(), any(), non_neg_integer()}
   def start_endpoint(endpoint, port, opts \\ []) do
-    opts = Keyword.validate!(opts, adapter: GRPC.Server.Adapters.Cowboy)
+    opts =
+      Keyword.validate!(opts,
+        adapter: GRPC.Server.Adapters.Cowboy,
+        adapter_opts: [],
+        exception_log_filter: nil,
+        max_body_size: nil
+      )
+
     adapter = opts[:adapter]
     servers = endpoint.__meta__(:servers)
     servers = GRPC.Server.servers_to_map(servers)

grpc_server/lib/grpc/server/adapters/cowboy/handler.ex

@@ -14,6 +14,10 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
   @default_trailers HTTP2.server_trailers()
   @trailers_flag 0b1000_0000
 
+  # 4 MB – matches gRPC-Go's default max receive message size.
+  # Override per-server with the :max_body_size option (bytes).
+  @default_max_body_size 4 * 1024 * 1024
+
   @type init_state :: {
           endpoint :: atom(),
           server :: {name :: String.t(), module()},
@@ -96,6 +100,8 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
           )
         end
 
+      max_body_size = Map.get(opts, :max_body_size, @default_max_body_size)
+
       {
         :cowboy_loop,
         req,
@@ -105,7 +111,8 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
           pending_reader: nil,
           access_mode: access_mode,
           codec: codec,
-          exception_log_filter: exception_log_filter
+          exception_log_filter: exception_log_filter,
+          max_body_size: max_body_size
         }
       }
     else
@@ -341,13 +348,19 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
   # APIs end
 
   def info({:read_full_body, ref, pid}, req, state) do
-    {s, body, req} = read_full_body(req, "", state[:handling_timer])
+    {s, body, req} = read_full_body(req, <<>>, state[:handling_timer], state.max_body_size)
     send(pid, {ref, {s, body}})
     {:ok, req, state}
   catch
     :exit, :timeout ->
       Logger.warning("Timeout when reading full body")
       info({:handling_timeout, self()}, req, state)
+
+    :throw, {:body_too_large, _size} ->
+      Logger.warning("Request body exceeded max_body_size (#{state.max_body_size} bytes)")
+      error = RPCError.exception(status: :resource_exhausted, message: "Request body too large")
+      req = send_error(req, error, state, :body_too_large)
+      {:stop, req, state}
   end
 
   def info({:read_body, ref, pid}, req, state) do
@@ -617,12 +630,27 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
     end
   end
 
-  defp read_full_body(req, body, timer) do
+  defp read_full_body(req, body, timer, max_bytes) do
     result = :cowboy_req.read_body(req, timeout_left_opt(timer))
 
     case result do
-      {:ok, data, req} -> {:ok, body <> data, req}
-      {:more, data, req} -> read_full_body(req, body <> data, timer)
+      {:ok, data, req} ->
+        total = body <> data
+
+        if byte_size(total) > max_bytes do
+          throw({:body_too_large, byte_size(total)})
+        else
+          {:ok, total, req}
+        end
+
+      {:more, data, req} ->
+        total = body <> data
+
+        if byte_size(total) > max_bytes do
+          throw({:body_too_large, byte_size(total)})
+        else
+          read_full_body(req, total, timer, max_bytes)
+        end
     end
   end
 
@@ -664,7 +692,10 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
   defp timeout_left_opt(timer, opts \\ %{}) do
     case timer do
       nil ->
-        Map.put(opts, :timeout, :infinity)
+        # No grpc-timeout header was supplied. Do not override cowboy's built-in
+        # per-chunk read timeout (15 s by default) with :infinity, which would
+        # allow a slow-trickle client to hold the connection open indefinitely.
+        opts
 
       timer ->
         case Process.read_timer(timer) do

grpc_server/mix.exs

@@ -41,6 +41,7 @@ defmodule GRPC.Server.MixProject do
       {:flow, "~> 1.2"},
       {:protobuf_generate, "~> 0.1.3", only: [:dev, :test]},
       {:ex_parameterized, "~> 1.3.7", only: :test},
+      {:gun, "~> 2.0", only: :test},
       {:mox, "~> 1.2", only: :test},
       {:ex_doc, "~> 0.39", only: [:dev, :docs], runtime: false},
       {:makeup, "~> 1.2.1", only: [:dev, :docs], runtime: false},

grpc_server/test/grpc/server/adapters/cowboy/handler_test.exs

@@ -0,0 +1,178 @@
+defmodule GRPC.Server.Adapters.Cowboy.HandlerTest do
+  use ExUnit.Case, async: false
+
+  import ExUnit.CaptureLog
+
+  # --------------------------------------------------------------------------
+  # Minimal server used across all tests
+  # --------------------------------------------------------------------------
+
+  defmodule HelloServer do
+    use GRPC.Server, service: Helloworld.Greeter.Service
+
+    def say_hello(req, _stream) do
+      %Helloworld.HelloReply{message: "Hello, #{req.name}"}
+    end
+  end
+
+  # --------------------------------------------------------------------------
+  # Helpers
+  # --------------------------------------------------------------------------
+
+  # Build a gRPC length-prefixed message frame (no compression).
+  defp grpc_frame(proto_binary) do
+    <<0::8, byte_size(proto_binary)::32, proto_binary::binary>>
+  end
+
+  defp grpc_request_headers do
+    [
+      {"content-type", "application/grpc+proto"},
+      {"te", "trailers"}
+    ]
+  end
+
+  # Open an HTTP/2 cleartext connection to the server and return the conn pid.
+  defp open_h2(port) do
+    {:ok, conn} = :gun.open(~c"localhost", port, %{protocols: [:http2]})
+    {:ok, :http2} = :gun.await_up(conn, 5_000)
+    conn
+  end
+
+  # Collect all gun frames for *stream_ref* until END_STREAM, then return the
+  # final grpc-status value found in either the response headers or trailers.
+  defp collect_grpc_status(conn, stream_ref) do
+    collect_grpc_status(conn, stream_ref, nil)
+  end
+
+  defp collect_grpc_status(conn, stream_ref, last_status) do
+    case :gun.await(conn, stream_ref, 5_000) do
+      {:response, :fin, _http_status, headers} ->
+        find_grpc_status(headers) || last_status
+
+      {:response, :nofin, _http_status, headers} ->
+        collect_grpc_status(conn, stream_ref, find_grpc_status(headers))
+
+      {:data, :fin, _data} ->
+        last_status
+
+      {:data, :nofin, _data} ->
+        collect_grpc_status(conn, stream_ref, last_status)
+
+      {:trailers, trailers} ->
+        find_grpc_status(trailers) || last_status
+
+      {:error, reason} ->
+        flunk("gun error: #{inspect(reason)}")
+    end
+  end
+
+  defp find_grpc_status(headers) do
+    case List.keyfind(headers, "grpc-status", 0) do
+      {"grpc-status", v} -> v
+      nil -> nil
+    end
+  end
+
+  # --------------------------------------------------------------------------
+  # Tests: max_body_size enforcement
+  # --------------------------------------------------------------------------
+
+  describe "max_body_size" do
+    test "rejects a body that exceeds max_body_size with RESOURCE_EXHAUSTED (8)" do
+      capture_log(fn ->
+        run_server_with_opts([HelloServer], [max_body_size: 64], fn port ->
+          # Build a gRPC frame whose total size is well above the 64-byte cap.
+          large_name = String.duplicate("x", 200)
+
+          body =
+            grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: large_name}))
+
+          assert byte_size(body) > 64,
+                 "test body (#{byte_size(body)} bytes) must exceed max_body_size: 64"
+
+          conn = open_h2(port)
+          ref = :gun.post(conn, "/helloworld.Greeter/SayHello", grpc_request_headers(), body)
+
+          assert collect_grpc_status(conn, ref) == "8"
+
+          :gun.close(conn)
+        end)
+      end)
+    end
+
+    test "allows a body within max_body_size and returns OK (0)" do
+      run_server_with_opts([HelloServer], [max_body_size: 4096], fn port ->
+        body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "hi"}))
+
+        assert byte_size(body) < 4096,
+               "test body (#{byte_size(body)} bytes) must fit within max_body_size: 4096"
+
+        conn = open_h2(port)
+        ref = :gun.post(conn, "/helloworld.Greeter/SayHello", grpc_request_headers(), body)
+
+        assert collect_grpc_status(conn, ref) == "0"
+
+        :gun.close(conn)
+      end)
+    end
+
+    test "default max_body_size is 4 MB – normal requests succeed without explicit option" do
+      run_server_with_opts([HelloServer], [], fn port ->
+        body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "default limit"}))
+
+        conn = open_h2(port)
+        ref = :gun.post(conn, "/helloworld.Greeter/SayHello", grpc_request_headers(), body)
+
+        assert collect_grpc_status(conn, ref) == "0"
+
+        :gun.close(conn)
+      end)
+    end
+  end
+
+  # --------------------------------------------------------------------------
+  # Tests: read timeout – no :infinity when grpc-timeout is absent
+  # --------------------------------------------------------------------------
+
+  describe "read timeout" do
+    test "omitting grpc-timeout header still completes a normal request" do
+      # If timeout_left_opt/1 incorrectly passed :infinity to cowboy for a
+      # nil timer, normal unary requests would still succeed – the regression
+      # is that a slow-trickle attack could hold the connection indefinitely.
+      # This smoke-test verifies the nil-timer path doesn't break normal calls.
+      run_server_with_opts([HelloServer], [], fn port ->
+        # Deliberately omit the grpc-timeout header.
+        headers = grpc_request_headers()
+        body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "no timeout header"}))
+
+        conn = open_h2(port)
+        ref = :gun.post(conn, "/helloworld.Greeter/SayHello", headers, body)
+
+        assert collect_grpc_status(conn, ref) == "0"
+
+        :gun.close(conn)
+      end)
+    end
+  end
+
+  # --------------------------------------------------------------------------
+  # Private helper: start a server with specific opts and run a test function
+  # --------------------------------------------------------------------------
+
+  defp run_server_with_opts(servers, opts, func) do
+    {:ok, _pid, port} =
+      start_supervised(%{
+        id: {GRPC.Server, System.unique_integer([:positive])},
+        start: {GRPC.Server, :start, [servers, 0, opts]},
+        type: :worker,
+        restart: :permanent,
+        shutdown: 500
+      })
+
+    try do
+      func.(port)
+    after
+      GRPC.Server.stop(servers)
+    end
+  end
+end

interop/script/run.exs

@@ -20,7 +20,8 @@ alias GRPC.Client.Adapters.Gun
 alias GRPC.Client.Adapters.Mint
 alias Interop.Client
 
-{:ok, _pid, port} = GRPC.Server.start_endpoint(Interop.Endpoint, port)
+# large_unary2! sends an 8 MB payload; allow up to 32 MB to keep headroom.
+{:ok, _pid, port} = GRPC.Server.start_endpoint(Interop.Endpoint, port, max_body_size: 32 * 1024 * 1024)
 
 defmodule InteropTestRunner do
   def run(_cli, adapter, port, rounds) do