fix(media-proxy): abort truncated streams

changelog.d/media-proxy-truncated-stream.fix

@@ -0,0 +1 @@
+Abort incomplete media proxy responses instead of caching partial bodies.

lib/pleroma/reverse_proxy.ex

@@ -12,7 +12,7 @@ defmodule Pleroma.ReverseProxy do
   @keep_resp_headers @resp_cache_headers ++
                        ~w(content-length content-type content-disposition content-encoding) ++
                        ~w(content-range accept-ranges vary)
-  @default_cache_control_header "public, max-age=1209600, immutable"
+  @default_cache_control_header "public, max-age=1209600, immutable, no-transform"
   @valid_resp_codes [200, 206, 304]
   @max_read_duration :timer.seconds(30)
   @max_body_length :infinity
@@ -24,6 +24,16 @@ defmodule Pleroma.ReverseProxy do
 
   @cachex Pleroma.Config.get([:cachex, :provider], Cachex)
 
+  defmodule StreamError do
+    @moduledoc false
+    defexception [:url, :reason]
+
+    @impl true
+    def message(%{url: url, reason: reason}) do
+      "reverse proxy stream from #{url} failed: #{inspect(reason)}"
+    end
+  end
+
   def max_read_duration_default, do: @max_read_duration
   def default_cache_control_header, do: @default_cache_control_header
 
@@ -190,34 +200,56 @@ defmodule Pleroma.ReverseProxy do
   end
 
   defp request_with_constraints(method, url, headers, client_opts, opts) do
-    case request(method, url, headers, client_opts) do
+    method
-      {:ok, _code, headers, client} = response ->
+    |> request(url, headers, client_opts)
-        case header_length_constraint(
+    |> constrain_response(Keyword.get(opts, :max_body_length, @max_body_length))
-               headers,
+  end
-               Keyword.get(opts, :max_body_length, @max_body_length)
-             ) do
-          :ok ->
-            response
 
-          error ->
+  defp constrain_response({:ok, code, headers, client}, limit) do
-            client().close(client)
+    case header_length_constraint(headers, limit) do
-            error
+      {:ok, headers} ->
-        end
+        {:ok, code, headers, client}
 
-      response ->
+      error ->
-        response
+        client().close(client)
+        error
     end
   end
 
+  defp constrain_response({:ok, code, headers}, limit) do
+    case header_length_constraint(headers, limit) do
+      {:ok, headers} -> {:ok, code, headers}
+      error -> error
+    end
+  end
+
+  defp constrain_response(response, _limit), do: response
+
   defp response(conn, client, url, status, headers, opts) do
     Logger.debug("#{__MODULE__} #{status} #{url} #{inspect(headers)}")
 
-    {headers, client, prefetched} = maybe_prefetch_for_content_type(headers, client, opts)
+    case maybe_prefetch_for_content_type(headers, client, opts) do
+      {_headers, client, {:error, {:upstream, error}}} ->
+        Logger.warning(
+          "#{__MODULE__} request to #{url} failed before streaming: #{inspect(error)}"
+        )
 
+        client().close(client)
+        track_failed_url(url, error, opts)
+
+        conn
+        |> error_or_redirect(url, 500, "Request failed", opts)
+        |> halt()
+
+      {headers, client, prefetched} ->
+        stream_response(conn, client, url, status, headers, opts, prefetched)
+    end
+  end
+
+  defp stream_response(conn, client, url, status, headers, opts, prefetched) do
     result =
       conn
       |> put_resp_headers(build_resp_headers(headers, opts))
-      |> streaming_compat
       |> send_chunked(status)
       |> chunk_reply(client, opts, prefetched)
 
@@ -225,17 +257,18 @@ defmodule Pleroma.ReverseProxy do
       {:ok, conn} ->
         halt(conn)
 
-      {:error, :closed, conn} ->
+      {:error, {:downstream, _error}, conn} ->
         client().close(client)
         halt(conn)
 
-      {:error, error, conn} ->
+      {:error, {:upstream, error}, conn} ->
         Logger.warning(
           "#{__MODULE__} request to #{url} failed while reading/chunking: #{inspect(error)}"
         )
 
         client().close(client)
-        halt(conn)
+        track_failed_url(url, error, opts)
+        raise_stream_error(conn, url, error)
     end
   end
 
@@ -252,10 +285,11 @@ defmodule Pleroma.ReverseProxy do
              byte_size(data),
              Keyword.get(opts, :max_body_length, @max_body_length)
            ),
-         {:ok, conn} <- chunk(conn, data) do
+         {:ok, conn} <- write_chunk(conn, data) do
       chunk_reply(conn, client, opts, byte_size(data), duration)
     else
-      {:error, error} -> {:error, error, conn}
+      {:error, {:downstream, error}} -> {:error, {:downstream, error}, conn}
+      {:error, error} -> {:error, {:upstream, error}, conn}
     end
   end
 
@@ -267,23 +301,37 @@ defmodule Pleroma.ReverseProxy do
              sent_so_far,
              Keyword.get(opts, :max_body_length, @max_body_length)
            ),
-         {:ok, conn} <- chunk(conn, data) do
+         {:ok, conn} <- write_chunk(conn, data) do
       chunk_reply(conn, client, opts, sent_so_far, duration)
     else
       :done -> {:ok, conn}
-      {:error, error} -> {:error, error, conn}
+      {:error, {source, error}} -> {:error, {source, error}, conn}
+      {:error, error} -> {:error, {:upstream, error}, conn}
     end
   end
 
   defp read_chunk(client, duration, opts) do
-    with {:ok, timer} <-
+    result =
-           check_read_duration(
+      with {:ok, timer} <-
-             duration,
+             check_read_duration(
-             Keyword.get(opts, :max_read_duration, @max_read_duration)
+               duration,
-           ),
+               Keyword.get(opts, :max_read_duration, @max_read_duration)
-         {:ok, data, client} <- client().stream_body(client),
+             ),
-         {:ok, duration} <- increase_read_duration(timer) do
+           {:ok, data, client} <- client().stream_body(client),
-      {:ok, data, client, duration}
+           {:ok, duration} <- increase_read_duration(timer) do
+        {:ok, data, client, duration}
+      end
+
+    case result do
+      {:error, error} -> {:error, {:upstream, error}}
+      result -> result
+    end
+  end
+
+  defp write_chunk(conn, data) do
+    case chunk(conn, data) do
+      {:error, error} -> {:error, {:downstream, error}}
+      result -> result
     end
   end
 
@@ -427,7 +475,24 @@ defmodule Pleroma.ReverseProxy do
     |> build_resp_cache_headers(opts)
     |> sanitise_content_type()
     |> build_resp_content_disposition_header(opts)
-    |> Keyword.merge(Keyword.get(opts, :resp_headers, []))
+    |> merge_headers(Keyword.get(opts, :resp_headers, []))
+    |> ensure_no_transform()
+  end
+
+  defp ensure_no_transform(headers) do
+    {_, cache_control} =
+      List.keyfind(headers, "cache-control", 0, {"cache-control", @default_cache_control_header})
+
+    directives =
+      cache_control
+      |> String.split(",")
+      |> Enum.map(&(&1 |> String.trim() |> String.downcase()))
+
+    if "no-transform" in directives do
+      headers
+    else
+      replace_header(headers, "cache-control", cache_control <> ", no-transform")
+    end
   end
 
   defp sanitise_content_type(headers) do
@@ -531,21 +596,34 @@ defmodule Pleroma.ReverseProxy do
     end
   end
 
-  defp header_length_constraint(headers, limit) when is_integer(limit) and limit > 0 do
+  defp header_length_constraint(headers, limit) do
-    with {_, size} <- List.keyfind(headers, "content-length", 0),
+    lengths =
-         {size, _} <- Integer.parse(size),
+      for {"content-length", value} <- headers,
-         true <- size <= limit do
+          do: parse_content_length(value)
-      :ok
+
-    else
+    case Enum.uniq(lengths) do
-      false ->
+      [] ->
-        {:error, :body_too_large}
+        {:ok, headers}
+
+      [size] when is_integer(size) ->
+        case body_size_constraint(size, limit) do
+          :ok -> {:ok, replace_header(headers, "content-length", to_string(size))}
+          error -> error
+        end
 
       _ ->
-        :ok
+        {:error, :invalid_content_length}
     end
   end
 
-  defp header_length_constraint(_, _), do: :ok
+  defp parse_content_length(value) when is_binary(value) do
+    case Integer.parse(value) do
+      {size, ""} when size >= 0 -> size
+      _ -> :invalid
+    end
+  end
+
+  defp parse_content_length(_value), do: :invalid
 
   defp body_size_constraint(size, limit) when is_integer(limit) and limit > 0 and size > limit do
     {:error, :body_too_large}
@@ -574,6 +652,40 @@ defmodule Pleroma.ReverseProxy do
 
   defp client, do: Pleroma.ReverseProxy.Client.Wrapper
 
+  # Neither Plug adapter exposes a public way to abort a committed response.
+  # Cowboy finalizes chunked HTTP/1 responses when the request process exits,
+  # so terminate the connection first to leave the response visibly incomplete.
+  defp raise_stream_error(
+         %Plug.Conn{adapter: {Plug.Cowboy.Conn, %{pid: connection_pid}}} = conn,
+         url,
+         error
+       ) do
+    if Plug.Conn.get_http_protocol(conn) in [:"HTTP/1.0", :"HTTP/1.1"] do
+      Process.exit(connection_pid, :kill)
+    end
+
+    raise StreamError, url: url, reason: error
+  end
+
+  # Bandit turns this exception into RST_STREAM; a regular exception sends an
+  # empty DATA frame with END_STREAM and makes the partial body look complete.
+  defp raise_stream_error(
+         %Plug.Conn{
+           adapter: {Bandit.Adapter, %{transport: %{stream_id: stream_id}}}
+         },
+         url,
+         error
+       ) do
+    raise Bandit.HTTP2.Errors.StreamError,
+      message: StreamError.message(%StreamError{url: url, reason: error}),
+      error_code: Bandit.HTTP2.Errors.internal_error(),
+      stream_id: stream_id
+  end
+
+  defp raise_stream_error(_conn, url, error) do
+    raise StreamError, url: url, reason: error
+  end
+
   defp track_failed_url(url, error, opts) do
     ttl =
       unless error in [:body_too_large, 400, 204] do
@@ -585,19 +697,6 @@ defmodule Pleroma.ReverseProxy do
     @cachex.put(:failed_proxy_url_cache, url, true, ttl: ttl)
   end
 
-  # When Cowboy handles a chunked response with a content-length header it streams
-  # over HTTP 1.1 instead of chunking. Bandit cannot stream over HTTP 1.1 so the header
-  # must be stripped or it breaks RFC compliance for Transfer Encoding: Chunked. RFC9112§6.2
-  #
-  # HTTP2 is always streamed for all adapters.
-  defp streaming_compat(conn) do
-    with Phoenix.Endpoint.Cowboy2Adapter <- Pleroma.Web.Endpoint.config(:adapter) do
-      conn
-    else
-      _ -> delete_resp_header(conn, "content-length")
-    end
-  end
-
   # Only when Tesla adapter is Hackney or Finch does the URL
   # need encoding before Reverse Proxying as both end up
   # using the raw Hackney client and cannot leverage our

test/pleroma/reverse_proxy_test.exs

@@ -11,6 +11,15 @@ defmodule Pleroma.ReverseProxyTest do
   alias Pleroma.ReverseProxy.ClientMock
   alias Plug.Conn
 
+  defmodule ProxyPlug do
+    @moduledoc false
+
+    def init(options), do: options
+
+    def call(conn, {url, opts}), do: Pleroma.ReverseProxy.call(conn, url, opts)
+    def call(conn, url), do: Pleroma.ReverseProxy.call(conn, url)
+  end
+
   setup_all do
     {:ok, _} = Registry.start_link(keys: :unique, name: ClientMock)
     :ok
@@ -18,6 +27,88 @@ defmodule Pleroma.ReverseProxyTest do
 
   setup :verify_on_exit!
 
+  defp start_bandit(url, opts \\ []) do
+    start_supervised!(
+      {Bandit,
+       plug: {ProxyPlug, {url, opts}},
+       ip: {127, 0, 0, 1},
+       port: 0,
+       http_options: [log_exceptions_with_status_codes: []]}
+    )
+  end
+
+  defp start_cowboy(url, opts \\ []) do
+    ref = {__MODULE__, make_ref()}
+
+    {:ok, _pid} =
+      Plug.Cowboy.http(ProxyPlug, {url, opts},
+        ip: {127, 0, 0, 1},
+        port: 0,
+        ref: ref,
+        protocol_options: [stream_handlers: [:cowboy_stream_h]]
+      )
+
+    on_exit(fn -> Plug.Cowboy.shutdown(ref) end)
+    :ranch.get_port(ref)
+  end
+
+  defp bandit_request(url, headers \\ [], opts \\ []) do
+    pid = start_bandit(url, opts)
+    {:ok, {{127, 0, 0, 1}, port}} = ThousandIsland.listener_info(pid)
+    http1_request(port, headers)
+  end
+
+  defp cowboy_request(url, headers \\ [], opts \\ []) do
+    url
+    |> start_cowboy(opts)
+    |> http1_request(headers)
+  end
+
+  defp http1_request(port, headers) do
+    {:ok, socket} = :gen_tcp.connect({127, 0, 0, 1}, port, [:binary, active: false])
+
+    request_headers =
+      [["host: localhost", "connection: close"] | headers]
+      |> List.flatten()
+      |> Enum.join("\r\n")
+
+    :ok = :gen_tcp.send(socket, "GET / HTTP/1.1\r\n#{request_headers}\r\n\r\n")
+    recv_all(socket, [])
+  end
+
+  defp recv_all(socket, chunks) do
+    case :gen_tcp.recv(socket, 0, 5_000) do
+      {:ok, data} -> recv_all(socket, [data | chunks])
+      {:error, reason} -> {reason, chunks |> Enum.reverse() |> IO.iodata_to_binary()}
+    end
+  end
+
+  defp receive_mint_responses(conn, responses \\ []) do
+    receive do
+      message ->
+        case Mint.HTTP2.stream(conn, message) do
+          {:ok, conn, new_responses} ->
+            responses = responses ++ new_responses
+
+            if Enum.any?(new_responses, fn response ->
+                 elem(response, 0) in [:done, :error]
+               end) do
+              responses
+            else
+              receive_mint_responses(conn, responses)
+            end
+
+          {:error, _conn, error, new_responses} ->
+            responses ++ new_responses ++ [{:connection_error, error}]
+
+          :unknown ->
+            receive_mint_responses(conn, responses)
+        end
+    after
+      2_000 -> responses ++ [:timeout]
+    end
+  end
+
   defp request_mock(invokes) do
     ClientMock
     |> expect(:request, fn :get, url, headers, _body, _opts ->
@@ -43,6 +134,45 @@ defmodule Pleroma.ReverseProxyTest do
     end)
   end
 
+  defp stream_then_error_mock(url, headers, chunk \\ "partial", error \\ :closed) do
+    ClientMock
+    |> expect(:request, fn :get, ^url, _, _, _ ->
+      Registry.register(ClientMock, url, 0)
+      {:ok, 200, headers, %{url: url}}
+    end)
+    |> expect(:stream_body, 2, fn %{url: ^url} = client ->
+      case Registry.lookup(ClientMock, url) do
+        [{_, 0}] ->
+          Registry.update_value(ClientMock, url, &(&1 + 1))
+          {:ok, chunk, client}
+
+        [{_, 1}] ->
+          Registry.unregister(ClientMock, url)
+          {:error, error}
+      end
+    end)
+    |> expect(:close, fn _ -> :ok end)
+  end
+
+  defp complete_stream_mock(url, headers, body) do
+    ClientMock
+    |> expect(:request, fn :get, ^url, _, _, _ ->
+      Registry.register(ClientMock, url, 0)
+      {:ok, 200, headers, %{url: url}}
+    end)
+    |> expect(:stream_body, 2, fn %{url: ^url} = client ->
+      case Registry.lookup(ClientMock, url) do
+        [{_, 0}] ->
+          Registry.update_value(ClientMock, url, &(&1 + 1))
+          {:ok, body, client}
+
+        [{_, 1}] ->
+          Registry.unregister(ClientMock, url)
+          :done
+      end
+    end)
+  end
+
   describe "reverse proxy" do
     test "do not track successful request", %{conn: conn} do
       request_mock(2)
@@ -70,14 +200,202 @@ defmodule Pleroma.ReverseProxyTest do
     assert response == %{"user-agent" => Pleroma.Application.user_agent()}
   end
 
-  test "closed connection", %{conn: conn} do
+  test "aborts a partially streamed response when the upstream closes", %{conn: conn} do
-    ClientMock
+    stream_then_error_mock("/closed", [{"content-length", "20"}])
-    |> expect(:request, fn :get, "/closed", _, _, _ -> {:ok, 200, [], %{}} end)
-    |> expect(:stream_body, fn _ -> {:error, :closed} end)
-    |> expect(:close, fn _ -> :ok end)
 
-    conn = ReverseProxy.call(conn, "/closed")
+    assert_raise ReverseProxy.StreamError, fn ->
-    assert conn.halted
+      ReverseProxy.call(conn, "/closed")
+    end
+
+    assert Cachex.get(:failed_proxy_url_cache, "/closed") == {:ok, true}
+  end
+
+  describe "Bandit HTTP/1 streaming" do
+    test "closes an incomplete chunked response without a terminating chunk" do
+      Mox.set_mox_global(ClientMock)
+      stream_then_error_mock("/bandit-closed", [{"content-type", "image/png"}])
+
+      assert {:closed, response} = bandit_request("/bandit-closed")
+      assert response =~ "transfer-encoding: chunked"
+      assert response =~ "partial"
+      refute response =~ "0\r\n\r\n"
+    end
+
+    test "closes an incomplete fixed-length response before the declared length" do
+      Mox.set_mox_global(ClientMock)
+
+      stream_then_error_mock("/bandit-fixed-closed", [
+        {"content-length", "20"},
+        {"content-type", "image/png"}
+      ])
+
+      assert {:closed, response} = bandit_request("/bandit-fixed-closed")
+      assert response =~ "content-length: 20"
+      refute response =~ "transfer-encoding"
+      assert String.ends_with?(response, "partial")
+    end
+
+    test "streams a complete fixed-length response without transformation" do
+      Mox.set_mox_global(ClientMock)
+      body = "complete body"
+
+      complete_stream_mock(
+        "/bandit-complete",
+        [
+          {"content-length", to_string(byte_size(body))},
+          {"content-type", "image/png"}
+        ],
+        body
+      )
+
+      assert {:closed, response} =
+               bandit_request("/bandit-complete", ["accept-encoding: deflate"])
+
+      assert response =~ "content-length: #{byte_size(body)}"
+      assert response =~ "cache-control: public, max-age=1209600, immutable, no-transform"
+      refute response =~ "content-encoding: gzip"
+      refute response =~ "content-encoding: deflate"
+      assert String.ends_with?(response, body)
+    end
+
+    test "keeps no-transform when custom response headers replace cache-control" do
+      Mox.set_mox_global(ClientMock)
+      body = "complete body"
+
+      complete_stream_mock(
+        "/bandit-custom-cache-control",
+        [
+          {"content-length", to_string(byte_size(body))},
+          {"content-type", "image/png"}
+        ],
+        body
+      )
+
+      assert {:closed, response} =
+               bandit_request(
+                 "/bandit-custom-cache-control",
+                 ["accept-encoding: deflate"],
+                 resp_headers: [{"cache-control", "private, max-age=60"}]
+               )
+
+      assert response =~ "content-length: #{byte_size(body)}"
+      assert response =~ "cache-control: private, max-age=60, no-transform"
+      refute response =~ "content-encoding: gzip"
+      refute response =~ "content-encoding: deflate"
+      assert String.ends_with?(response, body)
+    end
+
+    test "does not blacklist the upstream when the downstream disconnects" do
+      Mox.set_mox_global(ClientMock)
+      test_pid = self()
+
+      ClientMock
+      |> expect(:request, fn :get, "/bandit-client-closed", _, _, _ ->
+        {:ok, 200, [{"content-type", "image/png"}], %{url: "/bandit-client-closed"}}
+      end)
+      |> stub(:stream_body, fn client ->
+        send(test_pid, {:stream_requested, self()})
+
+        receive do
+          :continue -> {:ok, :binary.copy("x", 1_000_000), client}
+        after
+          1_000 -> raise "timed out waiting for downstream disconnect"
+        end
+      end)
+      |> expect(:close, fn _ ->
+        send(test_pid, :upstream_closed)
+        :ok
+      end)
+
+      pid = start_bandit("/bandit-client-closed")
+      {:ok, {{127, 0, 0, 1}, port}} = ThousandIsland.listener_info(pid)
+      {:ok, socket} = :gen_tcp.connect({127, 0, 0, 1}, port, [:binary, active: false])
+      :ok = :gen_tcp.send(socket, "GET / HTTP/1.1\r\nhost: localhost\r\n\r\n")
+
+      assert_receive {:stream_requested, request_pid}
+      :ok = :inet.setopts(socket, linger: {true, 0})
+      :ok = :gen_tcp.close(socket)
+      send(request_pid, :continue)
+
+      assert_receive :upstream_closed
+      assert Cachex.get(:failed_proxy_url_cache, "/bandit-client-closed") == {:ok, nil}
+    end
+  end
+
+  describe "Cowboy streaming" do
+    test "closes an incomplete HTTP/1 chunked response without a terminating chunk" do
+      Mox.set_mox_global(ClientMock)
+      stream_then_error_mock("/cowboy-closed", [{"content-type", "image/png"}])
+
+      assert {:closed, response} = cowboy_request("/cowboy-closed")
+      assert response =~ "transfer-encoding: chunked"
+      assert response =~ "partial"
+      refute response =~ "0\r\n\r\n"
+    end
+
+    test "closes an incomplete HTTP/1 fixed-length response before the declared length" do
+      Mox.set_mox_global(ClientMock)
+
+      stream_then_error_mock("/cowboy-fixed-closed", [
+        {"content-length", "20"},
+        {"content-type", "image/png"}
+      ])
+
+      assert {:closed, response} = cowboy_request("/cowboy-fixed-closed")
+      assert response =~ "content-length: 20"
+      refute response =~ "transfer-encoding"
+      assert String.ends_with?(response, "partial")
+    end
+
+    test "resets an incomplete HTTP/2 stream" do
+      Mox.set_mox_global(ClientMock)
+      stream_then_error_mock("/cowboy-http2-closed", [{"content-type", "image/png"}])
+
+      port = start_cowboy("/cowboy-http2-closed")
+      {:ok, mint} = Mint.HTTP2.connect(:http, "localhost", port)
+      {:ok, mint, request_ref} = Mint.HTTP2.request(mint, "GET", "/", [], nil)
+      responses = receive_mint_responses(mint)
+
+      assert {:status, request_ref, 200} in responses
+      assert {:data, request_ref, "partial"} in responses
+
+      assert Enum.any?(responses, fn
+               {:error, ^request_ref, %Mint.HTTPError{reason: {:server_closed_request, _}}} ->
+                 true
+
+               _ ->
+                 false
+             end)
+
+      refute {:done, request_ref} in responses
+      refute :timeout in responses
+    end
+  end
+
+  test "resets an incomplete Bandit HTTP/2 stream" do
+    Mox.set_mox_global(ClientMock)
+    stream_then_error_mock("/bandit-http2-closed", [{"content-type", "image/png"}])
+
+    pid = start_bandit("/bandit-http2-closed")
+    {:ok, {{127, 0, 0, 1}, port}} = ThousandIsland.listener_info(pid)
+    {:ok, mint} = Mint.HTTP2.connect(:http, "localhost", port)
+    {:ok, mint, request_ref} = Mint.HTTP2.request(mint, "GET", "/", [], nil)
+    responses = receive_mint_responses(mint)
+
+    assert {:status, request_ref, 200} in responses
+    assert {:data, request_ref, "partial"} in responses
+
+    assert Enum.any?(responses, fn
+             {:error, ^request_ref,
+              %Mint.HTTPError{reason: {:server_closed_request, :internal_error}}} ->
+               true
+
+             _ ->
+               false
+           end)
+
+    refute {:done, request_ref} in responses
+    refute :timeout in responses
   end
 
   defp stream_mock(invokes, with_close? \\ false) do
@@ -142,9 +460,14 @@ defmodule Pleroma.ReverseProxyTest do
     test "max_body_length returns error if streaming body more than that option", %{conn: conn} do
       stream_mock(3, true)
 
-      assert capture_log(fn ->
+      log =
-               ReverseProxy.call(conn, "/stream-bytes/50", max_body_length: 29)
+        capture_log(fn ->
-             end) =~
+          assert_raise ReverseProxy.StreamError, fn ->
+            ReverseProxy.call(conn, "/stream-bytes/50", max_body_length: 29)
+          end
+        end)
+
+      assert log =~
                "Elixir.Pleroma.ReverseProxy request to /stream-bytes/50 failed while reading/chunking: :body_too_large"
     end
 
@@ -169,6 +492,19 @@ defmodule Pleroma.ReverseProxyTest do
       assert Conn.get_resp_header(conn, "content-type") == ["image/png"]
       assert conn.resp_body == ""
     end
+
+    test "rejects a malformed content-length", %{conn: conn} do
+      url = "/head-invalid-content-length"
+
+      expect(ClientMock, :request, fn :head, ^url, _, _, _ ->
+        {:ok, 200, [{"content-length", "20junk"}]}
+      end)
+
+      conn = ReverseProxy.call(Map.put(conn, :method, "HEAD"), url)
+
+      assert conn.status == 500
+      assert Cachex.get(:failed_proxy_url_cache, url) == {:ok, true}
+    end
   end
 
   defp error_mock(status) when is_integer(status) do
@@ -271,6 +607,28 @@ defmodule Pleroma.ReverseProxyTest do
            ]
   end
 
+  test "redirects a content sniffing failure before committing the response", %{conn: conn} do
+    url = "https://example.com/extensionless"
+
+    ClientMock
+    |> expect(:request, fn :get, ^url, _, _, _ ->
+      {:ok, 200, [{"content-type", "application/octet-stream"}], %{url: url}}
+    end)
+    |> expect(:stream_body, fn _ -> {:error, :closed} end)
+    |> expect(:close, fn _ -> :ok end)
+
+    conn =
+      ReverseProxy.call(conn, url,
+        sniff_content_type: true,
+        redirect_on_failure: true
+      )
+
+    assert conn.status == 302
+    assert Conn.get_resp_header(conn, "location") == [url]
+    assert conn.state == :sent
+    assert Cachex.get(:failed_proxy_url_cache, url) == {:ok, true}
+  end
+
   test "preserves a generic content type for a non-image body", %{conn: conn} do
     body = "not an image"
 
@@ -365,7 +723,73 @@ defmodule Pleroma.ReverseProxyTest do
       |> expect(:stream_body, fn _ -> :done end)
 
       conn = ReverseProxy.call(conn, "/cache")
-      assert {"cache-control", "public, max-age=1209600, immutable"} in conn.resp_headers
+
+      assert {"cache-control", "public, max-age=1209600, immutable, no-transform"} in conn.resp_headers
+    end
+
+    test "preserves a valid upstream content-length", %{conn: conn} do
+      body = "complete body"
+
+      ClientMock
+      |> expect(:request, fn :get, "/content-length", _, _, _ ->
+        {:ok, 200, [{"content-length", to_string(byte_size(body))}], %{body: body}}
+      end)
+      |> expect(:stream_body, fn %{body: ^body} = client ->
+        {:ok, body, Map.delete(client, :body)}
+      end)
+      |> expect(:stream_body, fn %{} -> :done end)
+
+      conn = ReverseProxy.call(conn, "/content-length")
+
+      assert Conn.get_resp_header(conn, "content-length") == [to_string(byte_size(body))]
+    end
+
+    test "rejects malformed upstream content-length values" do
+      Enum.each(["20junk", "-1"], fn content_length ->
+        url = "/invalid-content-length/#{content_length}"
+
+        ClientMock
+        |> expect(:request, fn :get, ^url, _, _, _ ->
+          {:ok, 200, [{"content-length", content_length}], %{url: url}}
+        end)
+        |> expect(:close, fn %{url: ^url} -> :ok end)
+
+        conn = ReverseProxy.call(build_conn(), url)
+
+        assert conn.status == 500
+        assert Cachex.get(:failed_proxy_url_cache, url) == {:ok, true}
+      end)
+    end
+
+    test "rejects conflicting upstream content-length values" do
+      url = "/conflicting-content-length"
+
+      ClientMock
+      |> expect(:request, fn :get, ^url, _, _, _ ->
+        {:ok, 200, [{"content-length", "7"}, {"content-length", "20"}], %{url: url}}
+      end)
+      |> stub(:stream_body, fn _ -> :done end)
+      |> stub(:close, fn _ -> :ok end)
+
+      conn = ReverseProxy.call(build_conn(), url)
+
+      assert conn.status == 500
+      assert Cachex.get(:failed_proxy_url_cache, url) == {:ok, true}
+    end
+
+    test "normalizes equivalent upstream content-length values", %{conn: conn} do
+      body = "partial"
+
+      complete_stream_mock(
+        "/duplicate-content-length",
+        [{"content-length", "7"}, {"content-length", "07"}],
+        body
+      )
+
+      conn = ReverseProxy.call(conn, "/duplicate-content-length")
+
+      assert Conn.get_resp_header(conn, "content-length") == ["7"]
+      assert conn.resp_body == body
     end
   end