fix(http): harden Hackney request lifecycle

lib/pleroma/http/adapter_helper/hackney.ex

@@ -28,6 +28,8 @@ defmodule Pleroma.HTTP.AdapterHelper.Hackney do
     |> add_scheme_opts(uri)
     |> maybe_add_with_body()
     |> Pleroma.HTTP.AdapterHelper.maybe_add_proxy(proxy)
+    |> Keyword.put(:follow_redirect, false)
+    |> Keyword.put(:force_redirect, false)
   end
 
   defp add_scheme_opts(opts, _), do: opts

lib/pleroma/reverse_proxy.ex

@@ -98,7 +98,13 @@ defmodule Pleroma.ReverseProxy do
   def call(_conn, _url, _opts \\ [])
 
   def call(conn = %{method: method}, url, opts) when method in @methods do
-    client_opts = Keyword.merge(@default_options, Keyword.get(opts, :http, []))
+    client_opts =
+      @default_options
+      |> Keyword.merge(Keyword.get(opts, :http, []))
+      |> Keyword.put_new(
+        :recv_timeout,
+        receive_timeout(opts)
+      )
 
     req_headers = build_req_headers(conn.req_headers, opts)
 
@@ -423,6 +429,13 @@ defmodule Pleroma.ReverseProxy do
 
   defp check_read_duration(_, _), do: {:ok, :no_duration_limit, :no_duration_limit}
 
+  defp receive_timeout(opts) do
+    case Keyword.get(opts, :max_read_duration, @max_read_duration) do
+      timeout when is_integer(timeout) and timeout > 0 -> timeout
+      _ -> :infinity
+    end
+  end
+
   defp increase_read_duration({previous_duration, started})
        when is_integer(previous_duration) and is_integer(started) do
     duration = :erlang.system_time(:millisecond) - started

lib/pleroma/reverse_proxy/client/hackney.ex

@@ -60,7 +60,6 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do
     case receive_response(ref, recv_timeout) do
       {:ok, data} when is_binary(data) -> {:ok, data, client}
       {:ok, :done} -> :done
-      {:ok, {:error, error}} -> {:error, error}
       {:error, error} -> {:error, error}
     end
   end
@@ -86,21 +85,42 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do
       |> Keyword.put(:async, :once)
       |> Keyword.put(:stream_to, self())
 
-    with {:ok, ref} <- :hackney.request(method, url, headers, body, req_opts),
-         {: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}}
+    with {:ok, ref} <- :hackney.request(method, url, headers, body, req_opts) 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}}
+        end
+
+      case result do
+        {:ok, _, _, _} = response ->
+          response
+
+        error ->
+          close(ref)
+          flush_responses(ref)
+          error
+      end
     end
   end
 
   defp receive_response(ref, recv_timeout) do
     receive do
+      {:hackney_response, ^ref, {:error, error}} -> {:error, error}
       {:hackney_response, ^ref, response} -> {:ok, response}
     after
       recv_timeout -> {:error, :timeout}
     end
   end
 
+  defp flush_responses(ref) do
+    receive do
+      {:hackney_response, ^ref, _response} -> flush_responses(ref)
+    after
+      0 -> :ok
+    end
+  end
+
   defp redirect(url, resp_headers, env, limit) when limit == 0 do
     new_url = absolute_redirect_url(url, resp_headers)