diff --git a/changelog.d/reverse-proxy-client-leak.fix b/changelog.d/reverse-proxy-client-leak.fix new file mode 100644 index 000000000..e1ce4c967 --- /dev/null +++ b/changelog.d/reverse-proxy-client-leak.fix @@ -0,0 +1 @@ +Release the reverse proxy HTTP client on every outcome, fixing a memory leak where abandoned Hackney connection processes retained the whole response body diff --git a/lib/pleroma/reverse_proxy.ex b/lib/pleroma/reverse_proxy.ex index eaf7f1f51..55bcd13d3 100644 --- a/lib/pleroma/reverse_proxy.ex +++ b/lib/pleroma/reverse_proxy.ex @@ -118,7 +118,8 @@ defmodule Pleroma.ReverseProxy do with {:ok, nil} <- @cachex.get(:failed_proxy_url_cache, url), {:ok, code, headers, client} <- request(method, url, req_headers, client_opts), :ok <- - header_length_constraint( + enforce_header_length( + client, headers, Keyword.get(opts, :max_body_length, @max_body_length) ) do @@ -176,7 +177,8 @@ defmodule Pleroma.ReverseProxy do {:ok, code, headers} when code in @valid_resp_codes -> {:ok, code, downcase_headers(headers)} - {:ok, code, _, _} -> + {:ok, code, _, client} -> + client().close(client) {:error, {:invalid_http_response, code}} {:ok, code, _} -> @@ -197,12 +199,16 @@ defmodule Pleroma.ReverseProxy do |> send_chunked(status) |> chunk_reply(client, opts) + # Release the ref on every outcome, including a fully streamed body. The + # connection process behind it is not tied to this request and keeps the + # response body's binaries alive until it is explicitly closed. + client().close(client) + case result do {:ok, conn} -> halt(conn) {:error, :closed, conn} -> - client().close(client) halt(conn) {:error, error, conn} -> @@ -210,7 +216,6 @@ defmodule Pleroma.ReverseProxy do "#{__MODULE__} request to #{url} failed while reading/chunking: #{inspect(error)}" ) - client().close(client) halt(conn) end end @@ -396,6 +401,19 @@ defmodule Pleroma.ReverseProxy do end end + # Rejecting on content-length happens after the upstream response was already + # accepted, so the ref is live and has to be released before we bail out. + defp enforce_header_length(client, headers, limit) do + case header_length_constraint(headers, limit) do + :ok -> + :ok + + {:error, _} = error -> + client().close(client) + error + end + end + defp header_length_constraint(headers, limit) when is_integer(limit) and limit > 0 do with {_, size} <- List.keyfind(headers, "content-length", 0), {size, _} <- Integer.parse(size), diff --git a/lib/pleroma/reverse_proxy/client/hackney.ex b/lib/pleroma/reverse_proxy/client/hackney.ex index bf2979df9..c944fd38a 100644 --- a/lib/pleroma/reverse_proxy/client/hackney.ex +++ b/lib/pleroma/reverse_proxy/client/hackney.ex @@ -54,6 +54,8 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do end @impl true + def stream_body(%{released: true}), do: :done + def stream_body(%{ref: ref, recv_timeout: recv_timeout} = client) do _ = :hackney.stream_next(ref) @@ -73,6 +75,8 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do end @impl true + def close(%{released: true}), do: :ok + def close(%{ref: ref}), do: :hackney.close(ref) def close(ref), do: :hackney.close(ref) @@ -89,7 +93,7 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do result = with {:ok, {:status, code, _reason}} <- receive_response(ref, recv_timeout), {:ok, {:headers, resp_headers}} <- receive_response(ref, recv_timeout) do - {:ok, code, resp_headers, %{ref: ref, recv_timeout: recv_timeout}} + {:ok, code, resp_headers, build_client(ref, recv_timeout, method, code, opts)} end case result do @@ -104,6 +108,43 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do end end + # A response that cannot carry a body (204, 304, or a reply to HEAD) is + # complete as soon as the headers are parsed. Hackney queues `done` for us and, + # if the connection is pooled, checks it straight back in, so the ref stops + # being ours at that point: pulling the stream or stopping the connection + # afterwards would disturb whichever request checks it out next. Mark it so + # `stream_body/1` and `close/1` leave it alone. + # + # This relies on hackney never running its own redirect handling for us -- + # `request/5` forces `follow_redirect: false` -- otherwise a redirect status + # would be checked back in too and would need the same treatment. + defp build_client(ref, recv_timeout, method, code, opts) do + client = %{ref: ref, recv_timeout: recv_timeout} + + if released?(method, code) and pooled?(opts) do + flush_responses(ref) + Map.put(client, :released, true) + else + client + end + end + + defp released?(method, _code) when method in [:head, :HEAD, "head", "HEAD"], do: true + defp released?(_method, code) when code in [204, 304], do: true + defp released?(_method, _code), do: false + + # Mirrors hackney's own pooling decision: :hackney.use_pool/1, plus the fact + # that a proxy tunnel always bypasses the pool. An unpooled connection is never + # checked in, and parks in `connected` with no idle timeout, so it leaks + # forever unless we stop it ourselves. + defp pooled?(opts) do + is_nil(Keyword.get(opts, :proxy)) and use_pool?(Keyword.get(opts, :pool)) + end + + defp use_pool?(false), do: false + defp use_pool?(nil), do: Application.get_env(:hackney, :use_default_pool, true) + defp use_pool?(_pool_name), do: true + defp receive_response(ref, recv_timeout) do receive do {:hackney_response, ^ref, {:error, error}} -> {:error, error} diff --git a/lib/pleroma/reverse_proxy/client/tesla.ex b/lib/pleroma/reverse_proxy/client/tesla.ex index 4596d7a7f..b68940dc5 100644 --- a/lib/pleroma/reverse_proxy/client/tesla.ex +++ b/lib/pleroma/reverse_proxy/client/tesla.ex @@ -42,13 +42,14 @@ defmodule Pleroma.ReverseProxy.Client.Tesla do end end + # Reaching the end of the body does not release the ref: callers must always + # call close/1, exactly once, whatever the outcome. Releasing here as well + # would remove the client twice and crash the pool worker on the missing + # monitor. @impl true @spec stream_body(map()) :: {:ok, binary(), map()} | {:error, atom() | String.t()} | :done | no_return() - def stream_body(%{pid: pid, fin: true}) do - ConnectionPool.release_conn(pid) - :done - end + def stream_body(%{fin: true}), do: :done def stream_body(client) do case read_chunk!(client) do diff --git a/test/pleroma/http/hackney_follow_redirect_regression_test.exs b/test/pleroma/http/hackney_follow_redirect_regression_test.exs index a9f8a90b7..5082d080a 100644 --- a/test/pleroma/http/hackney_follow_redirect_regression_test.exs +++ b/test/pleroma/http/hackney_follow_redirect_regression_test.exs @@ -147,6 +147,87 @@ defmodule Pleroma.HTTP.HackneyFollowRedirectRegressionTest do Pleroma.ReverseProxy.Client.Hackney.request(:get, url, [], "", opts) end + # A pooled connection serving a bodyless response is checked back in by hackney + # at header time, so stopping that ref afterwards would evict a connection + # another request may already hold. These go direct rather than through the + # CONNECT proxy, because a proxy tunnel is never pooled. + test "hackney reverse proxy marks a pooled bodyless response as already released", + %{tls_server: tls_server} do + url = "#{tls_server.base_url}/no_content" + + assert {:ok, 204, _headers, client} = + Pleroma.ReverseProxy.Client.Hackney.request( + :get, + url, + [], + "", + direct_opts(tls_server) + ) + + assert %{released: true} = client + assert :done = Pleroma.ReverseProxy.Client.Hackney.stream_body(client) + assert :ok = Pleroma.ReverseProxy.Client.Hackney.close(client) + end + + test "hackney reverse proxy still owns a response that carries a body", + %{tls_server: tls_server} do + url = "#{tls_server.base_url}/final" + + assert {:ok, 200, _headers, client} = + Pleroma.ReverseProxy.Client.Hackney.request( + :get, + url, + [], + "", + direct_opts(tls_server) + ) + + refute Map.has_key?(client, :released) + assert "ok" == collect_body(client) + Pleroma.ReverseProxy.Client.Hackney.close(client) + end + + # An unpooled connection is never checked in and parks forever with no idle + # timeout, so it stays ours to stop even when the response has no body. + test "hackney reverse proxy still owns a bodyless response over a proxy tunnel", %{ + tls_server: tls_server, + proxy: proxy + } do + url = "#{tls_server.base_url}/no_content" + + opts = Keyword.put(direct_opts(tls_server), :proxy, proxy.proxy_url) + + assert {:ok, 204, _headers, client} = + Pleroma.ReverseProxy.Client.Hackney.request(:get, url, [], "", opts) + + refute Map.has_key?(client, :released) + Pleroma.ReverseProxy.Client.Hackney.close(client) + end + + test "hackney reverse proxy still owns a bodyless response when pooling is off", %{ + tls_server: tls_server + } do + url = "#{tls_server.base_url}/no_content" + + opts = Keyword.put(direct_opts(tls_server), :pool, false) + + assert {:ok, 204, _headers, client} = + Pleroma.ReverseProxy.Client.Hackney.request(:get, url, [], "", opts) + + refute Map.has_key?(client, :released) + Pleroma.ReverseProxy.Client.Hackney.close(client) + end + + defp direct_opts(tls_server) do + [ + pool: :media, + ssl_options: tls_client_options(tls_server), + connect_timeout: 1_000, + recv_timeout: 1_000, + follow_redirect: false + ] + end + defp collect_body(ref, acc \\ "") do case Pleroma.ReverseProxy.Client.Hackney.stream_body(ref) do :done -> acc @@ -216,6 +297,9 @@ defmodule Pleroma.HTTP.HackneyFollowRedirectRegressionTest do "/closed" -> :ok + "/no_content" -> + send_ssl_response(ssl_socket, 204, "No Content", [], "") + "/delayed" -> Process.sleep(100) send_ssl_response(ssl_socket, 200, "OK", [], "ok") diff --git a/test/pleroma/reverse_proxy_test.exs b/test/pleroma/reverse_proxy_test.exs index 3c04f5078..20d65d283 100644 --- a/test/pleroma/reverse_proxy_test.exs +++ b/test/pleroma/reverse_proxy_test.exs @@ -18,6 +18,21 @@ defmodule Pleroma.ReverseProxyTest do setup :verify_on_exit! + # Every path that obtains a client ref releases it, so tests that do not assert + # on the release still need close/1 to be callable. Recording each call lets + # the tests below pin *exactly* once: a Mox `expect` would silently fall + # through to this stub on a second call. + setup do + test_pid = self() + + stub(ClientMock, :close, fn client -> + send(test_pid, {:closed, client}) + :ok + end) + + :ok + end + defp request_mock(invokes) do ClientMock |> expect(:request, fn :get, url, headers, _body, _opts -> @@ -91,6 +106,81 @@ defmodule Pleroma.ReverseProxyTest do end end + # The connection behind a client ref is not tied to the lifetime of this + # request. Hackney 4 in particular parks a non-pooled `hackney_conn` process + # in the `closed` state forever when nobody stops it, pinning every refc + # binary it read, so abandoning a ref leaks the whole response body. + describe "client ref release" do + test "closes the client after the body is fully streamed", %{conn: conn} do + ClientMock + |> expect(:request, fn :get, "/release", _, _, _ -> + {:ok, 200, [{"content-type", "image/png"}], %{url: "/release"}} + end) + |> expect(:stream_body, fn _ -> :done end) + + conn = ReverseProxy.call(conn, "/release") + + assert conn.halted + assert_received {:closed, %{url: "/release"}} + refute_received {:closed, _} + end + + test "closes the client when content-length exceeds max_body_length", %{conn: conn} do + ClientMock + |> expect(:request, fn :get, "/release-huge", _, _, _ -> + {:ok, 200, [{"content-length", "102227307"}], %{url: "/release-huge"}} + end) + + assert capture_log(fn -> + ReverseProxy.call(conn, "/release-huge", max_body_length: 25 * 1_048_576) + end) =~ ":body_too_large" + + assert_received {:closed, %{url: "/release-huge"}} + refute_received {:closed, _} + end + + test "closes the client when the upstream status is not proxiable", %{conn: conn} do + ClientMock + |> expect(:request, fn :get, "/release-418", _, _, _ -> + {:ok, 418, [], %{url: "/release-418"}} + end) + + assert capture_log(fn -> ReverseProxy.call(conn, "/release-418") end) =~ + "failed with HTTP status 418" + + assert_received {:closed, %{url: "/release-418"}} + refute_received {:closed, _} + end + + # The hackney client answers HEAD with a ref too, so this path must release + # it as well; only the ref-less 3-tuple reply has nothing to close. + test "closes the client when a HEAD response comes back with a ref", %{conn: conn} do + ClientMock + |> expect(:request, fn :head, "/release-head", _, _, _ -> + {:ok, 200, [{"content-type", "image/png"}], %{url: "/release-head"}} + end) + |> expect(:stream_body, fn _ -> :done end) + + conn = ReverseProxy.call(Map.put(conn, :method, "HEAD"), "/release-head") + + assert conn.halted + assert_received {:closed, %{url: "/release-head"}} + refute_received {:closed, _} + end + + test "has nothing to close when the reply carries no ref", %{conn: conn} do + ClientMock + |> expect(:request, fn :head, "/release-headless", _, _, _ -> + {:ok, 200, [{"content-type", "image/png"}]} + end) + + conn = ReverseProxy.call(Map.put(conn, :method, "HEAD"), "/release-headless") + + assert conn.status == 200 + refute_received {:closed, _} + end + end + test "use Pleroma's user agent in the request; don't pass the client's", %{conn: conn} do request_mock(2) @@ -116,7 +206,7 @@ defmodule Pleroma.ReverseProxyTest do assert conn.halted end - defp stream_mock(invokes, with_close? \\ false) do + defp stream_mock(invokes) do ClientMock |> expect(:request, fn :get, "/stream-bytes/" <> length, _, _, _ -> Registry.register(ClientMock, "/stream-bytes/" <> length, 0) @@ -142,10 +232,6 @@ defmodule Pleroma.ReverseProxyTest do :done end end) - - if with_close? do - expect(ClientMock, :close, fn _ -> :ok end) - end end describe "max_body" do @@ -165,12 +251,15 @@ defmodule Pleroma.ReverseProxyTest do end test "max_body_length returns error if streaming body more than that option", %{conn: conn} do - stream_mock(3, true) + stream_mock(3) assert capture_log(fn -> ReverseProxy.call(conn, "/stream-bytes/50", max_body_length: 30) end) =~ "Elixir.Pleroma.ReverseProxy request to /stream-bytes/50 failed while reading/chunking: :body_too_large" + + assert_received {:closed, %{url: "/stream-bytes/50"}} + refute_received {:closed, _} end end