basic support for proxies

lib/pleroma/gun/api/api.ex

@@ -6,20 +6,21 @@ defmodule Pleroma.Gun.API do
   @callback open(charlist(), pos_integer(), map()) :: {:ok, pid()}
   @callback info(pid()) :: map()
   @callback close(pid()) :: :ok
+  @callback await_up(pid) :: {:ok, atom()} | {:error, atom()}
+  @callback connect(pid(), map()) :: reference()
+  @callback await(pid(), reference()) :: {:response, :fin, 200, []}
 
-  def open(host, port, opts) do
-    api().open(host, port, opts)
-  end
+  def open(host, port, opts), do: api().open(host, port, opts)
 
-  def info(pid) do
-    api().info(pid)
-  end
+  def info(pid), do: api().info(pid)
 
-  def close(pid) do
-    api().close(pid)
-  end
+  def close(pid), do: api().close(pid)
 
-  defp api do
-    Pleroma.Config.get([Pleroma.Gun.API], Pleroma.Gun.API.Gun)
-  end
+  def await_up(pid), do: api().await_up(pid)
+
+  def connect(pid, opts), do: api().connect(pid, opts)
+
+  def await(pid, ref), do: api().await(pid, ref)
+
+  defp api, do: Pleroma.Config.get([Pleroma.Gun.API], Pleroma.Gun.API.Gun)
 end

lib/pleroma/gun/api/gun.ex

@@ -31,4 +31,13 @@ defmodule Pleroma.Gun.API.Gun do
 
   @impl API
   def close(pid), do: :gun.close(pid)
+
+  @impl API
+  def await_up(pid), do: :gun.await_up(pid)
+
+  @impl API
+  def connect(pid, opts), do: :gun.connect(pid, opts)
+
+  @impl API
+  def await(pid, ref), do: :gun.await(pid, ref)
 end

lib/pleroma/gun/api/mock.ex

@@ -73,6 +73,41 @@ defmodule Pleroma.Gun.API.Mock do
   end
 
   @impl API
+  def open({127, 0, 0, 1}, 8123, _) do
+    Task.start_link(fn -> Process.sleep(1_000) end)
+  end
+
+  @impl API
+  def open('localhost', 9050, _) do
+    Task.start_link(fn -> Process.sleep(1_000) end)
+  end
+
+  @impl API
+  def await_up(_pid) do
+    {:ok, :http}
+  end
+
+  @impl API
+  def connect(pid, %{host: _, port: 80}) do
+    ref = make_ref()
+    Registry.register(API.Mock, ref, pid)
+    ref
+  end
+
+  @impl API
+  def connect(pid, %{host: _, port: 443, protocols: [:http2], transport: :tls}) do
+    ref = make_ref()
+    Registry.register(API.Mock, ref, pid)
+    ref
+  end
+
+  @impl API
+  def await(pid, ref) do
+    [{_, ^pid}] = Registry.lookup(API.Mock, ref)
+    {:response, :fin, 200, []}
+  end
+
+  @impl API
   def info(pid) do
     [{_, info}] = Registry.lookup(API.Mock, pid)
     info

lib/pleroma/gun/connections.ex

@@ -4,6 +4,7 @@
 
 defmodule Pleroma.Gun.Connections do
   use GenServer
+  require Logger
 
   @type domain :: String.t()
   @type conn :: Pleroma.Gun.Conn.t()
@@ -154,14 +155,69 @@ defmodule Pleroma.Gun.Connections do
   end
 
   defp open_conn(key, uri, from, state, opts) do
-    {:ok, conn} = API.open(to_charlist(uri.host), uri.port, opts)
+    host = to_charlist(uri.host)
+    port = uri.port
 
-    state =
-      put_in(state.conns[key], %Pleroma.Gun.Conn{
-        conn: conn,
-        waiting_pids: [from]
-      })
+    result =
+      if opts[:proxy] do
+        with {proxy_host, proxy_port} <- opts[:proxy],
+             {:ok, conn} <- API.open(proxy_host, proxy_port, opts),
+             {:ok, _} <- API.await_up(conn) do
+          connect_opts = %{host: host, port: port}
 
-    {:noreply, state}
+          connect_opts =
+            if uri.scheme == "https" do
+              Map.put(connect_opts, :protocols, [:http2])
+              |> Map.put(:transport, :tls)
+            else
+              connect_opts
+            end
+
+          with stream <- API.connect(conn, connect_opts),
+               {:response, :fin, 200, _} <- API.await(conn, stream) do
+            {:ok, conn, true}
+          end
+        else
+          {:error, error} ->
+            {:error, error}
+
+          error ->
+            Logger.warn(inspect(error))
+            {:error, :error_connection_to_proxy}
+        end
+      else
+        with {:ok, conn} <- API.open(host, port, opts) do
+          {:ok, conn, false}
+        else
+          {:error, error} ->
+            {:error, error}
+
+          error ->
+            Logger.warn(inspect(error))
+            {:error, :error_connection}
+        end
+      end
+
+    case result do
+      {:ok, conn, is_up} ->
+        {from_list, used, conn_state} = if is_up, do: {[], 1, :up}, else: {[from], 0, :open}
+
+        state =
+          put_in(state.conns[key], %Pleroma.Gun.Conn{
+            conn: conn,
+            waiting_pids: from_list,
+            used: used,
+            state: conn_state
+          })
+
+        if is_up do
+          {:reply, conn, state}
+        else
+          {:noreply, state}
+        end
+
+      {:error, _error} ->
+        {:reply, nil, state}
+    end
   end
 end

lib/pleroma/http/connection.ex

@@ -14,6 +14,8 @@ defmodule Pleroma.HTTP.Connection do
     version: :master
   ]
 
+  require Logger
+
   @doc """
   Configure a client connection
 
@@ -33,13 +35,20 @@ defmodule Pleroma.HTTP.Connection do
   def options(opts) do
     options = Keyword.get(opts, :adapter, [])
     adapter_options = Pleroma.Config.get([:http, :adapter], [])
+
     proxy_url = Pleroma.Config.get([:http, :proxy_url], nil)
 
+    proxy =
+      case parse_proxy(proxy_url) do
+        {:ok, proxy_host, proxy_port} -> {proxy_host, proxy_port}
+        _ -> nil
+      end
+
     options =
       @options
       |> Keyword.merge(adapter_options)
       |> Keyword.merge(options)
-      |> Keyword.merge(proxy: proxy_url)
+      |> Keyword.merge(proxy: proxy)
 
     pool = options[:pool]
     url = options[:url]
@@ -75,4 +84,49 @@ defmodule Pleroma.HTTP.Connection do
         |> Keyword.put(:tls_opts, tls_opts)
     end
   end
+
+  @spec parse_proxy(String.t() | tuple() | nil) ::
+          {tuple, pos_integer()} | {:error, atom()} | nil
+  def parse_proxy(nil), do: nil
+
+  def parse_proxy(proxy) when is_binary(proxy) do
+    with [host, port] <- String.split(proxy, ":"),
+         {port, ""} <- Integer.parse(port) do
+      {:ok, parse_host(host), port}
+    else
+      {_, _} ->
+        Logger.warn("parsing port in proxy fail #{inspect(proxy)}")
+        {:error, :error_parsing_port_in_proxy}
+
+      :error ->
+        Logger.warn("parsing port in proxy fail #{inspect(proxy)}")
+        {:error, :error_parsing_port_in_proxy}
+
+      _ ->
+        Logger.warn("parsing proxy fail #{inspect(proxy)}")
+        {:error, :error_parsing_proxy}
+    end
+  end
+
+  def parse_proxy(proxy) when is_tuple(proxy) do
+    with {_type, host, port} <- proxy do
+      {:ok, parse_host(host), port}
+    else
+      _ ->
+        Logger.warn("parsing proxy fail #{inspect(proxy)}")
+        {:error, :error_parsing_proxy}
+    end
+  end
+
+  @spec parse_host(String.t() | tuple()) :: charlist() | atom()
+  def parse_host(host) when is_atom(host), do: to_charlist(host)
+
+  def parse_host(host) when is_binary(host) do
+    host = to_charlist(host)
+
+    case :inet.parse_address(host) do
+      {:error, :einval} -> host
+      {:ok, ip} -> ip
+    end
+  end
 end

test/gun/connections_test.exs

@@ -315,4 +315,126 @@ defmodule Gun.ConnectionsTest do
       } = Connections.get_state(name)
     end
   end
+
+  describe "with proxy usage" do
+    test "proxy as ip", %{name: name, pid: pid} do
+      conn =
+        Connections.get_conn(
+          "http://proxy_string.com",
+          [genserver_pid: pid, proxy: {{127, 0, 0, 1}, 8123}],
+          name
+        )
+
+      %Connections{
+        conns: %{
+          "http:proxy_string.com:80" => %Conn{
+            conn: ^conn,
+            state: :up,
+            waiting_pids: [],
+            used: 1
+          }
+        },
+        opts: [max_connections: 2, timeout: 10]
+      } = Connections.get_state(name)
+
+      reused_conn =
+        Connections.get_conn(
+          "http://proxy_string.com",
+          [genserver_pid: pid, proxy: {{127, 0, 0, 1}, 8123}],
+          name
+        )
+
+      assert reused_conn == conn
+    end
+
+    test "proxy as host", %{name: name, pid: pid} do
+      conn =
+        Connections.get_conn(
+          "http://proxy_tuple_atom.com",
+          [genserver_pid: pid, proxy: {'localhost', 9050}],
+          name
+        )
+
+      %Connections{
+        conns: %{
+          "http:proxy_tuple_atom.com:80" => %Conn{
+            conn: ^conn,
+            state: :up,
+            waiting_pids: [],
+            used: 1
+          }
+        },
+        opts: [max_connections: 2, timeout: 10]
+      } = Connections.get_state(name)
+
+      reused_conn =
+        Connections.get_conn(
+          "http://proxy_tuple_atom.com",
+          [genserver_pid: pid, proxy: {'localhost', 9050}],
+          name
+        )
+
+      assert reused_conn == conn
+    end
+
+    test "proxy as ip and ssl", %{name: name, pid: pid} do
+      conn =
+        Connections.get_conn(
+          "https://proxy_string.com",
+          [genserver_pid: pid, proxy: {{127, 0, 0, 1}, 8123}],
+          name
+        )
+
+      %Connections{
+        conns: %{
+          "https:proxy_string.com:443" => %Conn{
+            conn: ^conn,
+            state: :up,
+            waiting_pids: [],
+            used: 1
+          }
+        },
+        opts: [max_connections: 2, timeout: 10]
+      } = Connections.get_state(name)
+
+      reused_conn =
+        Connections.get_conn(
+          "https://proxy_string.com",
+          [genserver_pid: pid, proxy: {{127, 0, 0, 1}, 8123}],
+          name
+        )
+
+      assert reused_conn == conn
+    end
+
+    test "proxy as host and ssl", %{name: name, pid: pid} do
+      conn =
+        Connections.get_conn(
+          "https://proxy_tuple_atom.com",
+          [genserver_pid: pid, proxy: {'localhost', 9050}],
+          name
+        )
+
+      %Connections{
+        conns: %{
+          "https:proxy_tuple_atom.com:443" => %Conn{
+            conn: ^conn,
+            state: :up,
+            waiting_pids: [],
+            used: 1
+          }
+        },
+        opts: [max_connections: 2, timeout: 10]
+      } = Connections.get_state(name)
+
+      reused_conn =
+        Connections.get_conn(
+          "https://proxy_tuple_atom.com",
+          [genserver_pid: pid, proxy: {'localhost', 9050}],
+          name
+        )
+
+      assert reused_conn == conn
+    end
+  end
 end

test/http/connection_test.exs

@@ -0,0 +1,65 @@
+defmodule Pleroma.HTTP.ConnectionTest do
+  use ExUnit.Case, async: true
+  import ExUnit.CaptureLog
+  alias Pleroma.HTTP.Connection
+
+  describe "parse_host/1" do
+    test "as atom" do
+      assert Connection.parse_host(:localhost) == 'localhost'
+    end
+
+    test "as string" do
+      assert Connection.parse_host("localhost.com") == 'localhost.com'
+    end
+
+    test "as string ip" do
+      assert Connection.parse_host("127.0.0.1") == {127, 0, 0, 1}
+    end
+  end
+
+  describe "parse_proxy/1" do
+    test "ip with port" do
+      assert Connection.parse_proxy("127.0.0.1:8123") == {:ok, {127, 0, 0, 1}, 8123}
+    end
+
+    test "host with port" do
+      assert Connection.parse_proxy("localhost:8123") == {:ok, 'localhost', 8123}
+    end
+
+    test "as tuple" do
+      assert Connection.parse_proxy({:socks5, :localhost, 9050}) == {:ok, 'localhost', 9050}
+    end
+
+    test "as tuple with string host" do
+      assert Connection.parse_proxy({:socks5, "localhost", 9050}) == {:ok, 'localhost', 9050}
+    end
+
+    test "ip without port" do
+      capture_log(fn ->
+        assert Connection.parse_proxy("127.0.0.1") == {:error, :error_parsing_proxy}
+      end) =~ "parsing proxy fail \"127.0.0.1\""
+    end
+
+    test "host without port" do
+      capture_log(fn ->
+        assert Connection.parse_proxy("localhost") == {:error, :error_parsing_proxy}
+      end) =~ "parsing proxy fail \"localhost\""
+    end
+
+    test "host with bad port" do
+      capture_log(fn ->
+        assert Connection.parse_proxy("localhost:port") == {:error, :error_parsing_port_in_proxy}
+      end) =~ "parsing port in proxy fail \"localhost:port\""
+    end
+
+    test "as tuple without port" do
+      capture_log(fn ->
+        assert Connection.parse_proxy({:socks5, :localhost}) == {:error, :error_parsing_proxy}
+      end) =~ "parsing proxy fail {:socks5, :localhost}"
+    end
+
+    test "with nil" do
+      assert Connection.parse_proxy(nil) == nil
+    end
+  end
+end