No OneTemporary
Actions

Authored By

Unknown

Size

55 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/DEVLOG.md b/DEVLOG.md
	index 25f07171f..1e5d2b63d 100644
	--- a/DEVLOG.md
	+++ b/DEVLOG.md
	@@ -1,13 +1,14 @@
	# Development Log

	## Gun pool validation

	- The previous unique origin registry forced concurrent HTTP/1 requests onto one connection.
	- Pool identity omitted proxy routes and credentials, allowing incompatible routes to reuse a connection.
	- Gun 2.2 CONNECT requests require their tunnel stream reference on subsequent requests; Tesla's bundled Gun adapter does not preserve it.
	- CONNECT and SOCKS5 setup returned before the destination tunnel was ready and did not pass authentication credentials.
	- The existing global `max_connections` limit and idle reclaimer bound HTTP/1 sibling and many-host connection growth.
	- SOCKS4 is not supported by Gun 2.2 and is now rejected explicitly by the Gun adapter.
	- Gun reports a stale stream error to the process that calls `cancel/2`; pool workers must consume that reply so one HTTP/2 cancellation cannot terminate other multiplexed requests.
	- Streamed uploads must be registered before body enumeration and cancelled if enumeration exits, throws, or raises, otherwise a partial HTTP/1 request can remain reusable.
	- Gun 2.4 bounds HTTP/1 response header and trailer buffers, rejects invalid request headers by default, and fixes cross-origin HTTP/2 push handling; Gun 2.4.1 includes Cowlib 2.17.1 to repair HTTP/2 HPACK decoding.
	+- Plain HTTP proxy requests must use absolute-form request targets; CONNECT is reserved for HTTPS tunnels because forwarding proxies commonly reject CONNECT to port 80.
	diff --git a/changelog.d/gun-pool-rework.fix b/changelog.d/gun-pool-rework.fix
	index b7d9adccf..d555b5bce 100644
	--- a/changelog.d/gun-pool-rework.fix
	+++ b/changelog.d/gun-pool-rework.fix
	@@ -1 +1 @@
	-Fixed Gun HTTP/1 concurrency, authenticated proxy handling, tunnel requests, and streamed connection cleanup. Updated Gun and Cowlib for HTTP client security fixes.
	+Fixed Gun HTTP/1 concurrency, forwarding and authenticated proxy handling, tunnel requests, and streamed connection cleanup. Updated Gun and Cowlib for HTTP client security fixes.
	diff --git a/docs/configuration/howto_proxy.md b/docs/configuration/howto_proxy.md
	index 9e571d6fc..d805ff965 100644
	--- a/docs/configuration/howto_proxy.md
	+++ b/docs/configuration/howto_proxy.md
	@@ -1,14 +1,15 @@
	# How to configure upstream proxy for federation
	If you want to proxify all http requests (e.g. for TOR) that pleroma makes to an upstream proxy server, edit you config file (`dev.secret.exs` or `prod.secret.exs`) and add the following:

	```
	config :pleroma, :http,
	proxy_url: "127.0.0.1:8123"
	```

	The other way to do it, for example, with Tor you would most likely add something like this:
	```
	config :pleroma, :http, proxy_url: {:socks5, :localhost, 9050}
	```

	The Gun HTTP adapter supports SOCKS5. SOCKS4 proxies require another HTTP adapter.
	+For regular HTTP proxies, Gun forwards plain HTTP requests directly and uses CONNECT tunnels for HTTPS requests.
	diff --git a/lib/pleroma/gun/conn.ex b/lib/pleroma/gun/conn.ex
	index f56a1af2f..85a2fb0e8 100644
	--- a/lib/pleroma/gun/conn.ex
	+++ b/lib/pleroma/gun/conn.ex
	@@ -1,211 +1,226 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Gun.Conn do
	alias Pleroma.Gun

	require Logger

	def open(%URI{} = uri, opts) do
	pool_opts = Pleroma.Config.get([:connections_pool], [])

	opts =
	opts
	\|> Enum.into(%{})
	\|> Map.put_new(:connect_timeout, pool_opts[:connect_timeout] \|\| 5_000)
	\|> Map.put_new(:supervise, false)
	\|> maybe_add_tls_opts(uri)

	do_open(uri, opts)
	end

	defp maybe_add_tls_opts(opts, %URI{scheme: "http"}), do: opts

	defp maybe_add_tls_opts(opts, %URI{scheme: "https"}) do
	tls_opts = [
	verify: :verify_peer,
	cacertfile: CAStore.file_path(),
	depth: 20,
	reuse_sessions: false,
	log_level: :warning,
	customize_hostname_check: [match_fun: :public_key.pkix_verify_hostname_match_fun(:https)]
	]

	tls_opts =
	if Keyword.keyword?(opts[:tls_opts]) do
	Keyword.merge(tls_opts, opts[:tls_opts])
	else
	tls_opts
	end

	Map.put(opts, :tls_opts, tls_opts)
	end

	+ defp do_open(%URI{scheme: "http"} = uri, %{proxy: {proxy_host, proxy_port}} = opts) do
	+ with open_opts <- opts \|> proxy_open_opts() \|> Map.put(:protocols, [:http]),
	+ {:ok, conn} <- Gun.open(proxy_host, proxy_port, open_opts),
	+ {:ok, protocol} <- await_up(conn, opts[:connect_timeout]) do
	+ {:ok, conn, protocol, :forward_proxy}
	+ else
	+ error ->
	+ Logger.warning(
	+ "Opening forward proxy connection to #{compose_uri_log(uri)} failed with error #{inspect(error)}"
	+ )
	+
	+ error
	+ end
	+ end
	+
	defp do_open(uri, %{proxy: {proxy_host, proxy_port}} = opts) do
	connect_opts =
	uri
	\|> destination_opts()
	\|> add_http2_opts(uri.scheme, Map.get(opts, :tls_opts, []))
	\|> add_proxy_auth(opts)

	with open_opts <- proxy_open_opts(opts),
	{:ok, conn} <- Gun.open(proxy_host, proxy_port, open_opts),
	{:ok, _proxy_protocol} <- await_up(conn, opts[:connect_timeout]) do
	conn
	\|> Gun.connect(connect_opts)
	\|> await_proxy_connect(conn, opts[:connect_timeout])
	else
	error ->
	Logger.warning(
	"Opening proxied connection to #{compose_uri_log(uri)} failed with error #{inspect(error)}"
	)

	error
	end
	end

	defp do_open(_uri, %{proxy: {:socks4, _proxy_host, _proxy_port}}) do
	{:error, :socks4_unsupported}
	end

	defp do_open(uri, %{proxy: {proxy_type, proxy_host, proxy_port}} = opts)
	when proxy_type in [:socks, :socks5] do
	socks_opts =
	uri
	\|> destination_opts()
	\|> add_http2_opts(uri.scheme, Map.get(opts, :tls_opts, []))
	\|> Map.put(:version, 5)
	\|> add_socks_proxy_auth(opts)

	open_opts =
	opts
	\|> proxy_open_opts()
	\|> Map.put(:protocols, [:socks])
	\|> Map.put(:socks_opts, socks_opts)

	with {:ok, conn} <- Gun.open(proxy_host, proxy_port, open_opts),
	{:ok, :socks} <- await_up(conn, opts[:connect_timeout]),
	{:ok, protocol} <- await_tunnel_up(conn, :undefined, opts[:connect_timeout]) do
	{:ok, conn, protocol, nil}
	else
	error ->
	Logger.warning(
	"Opening socks proxied connection to #{compose_uri_log(uri)} failed with error #{inspect(error)}"
	)

	error
	end
	end

	defp do_open(%URI{host: host, port: port} = uri, opts) do
	host = Pleroma.HTTP.AdapterHelper.parse_host(host)
	opts = Map.put(opts, :transport, transport(uri.scheme))

	with {:ok, conn} <- Gun.open(host, port, opts),
	{:ok, protocol} <- await_up(conn, opts[:connect_timeout]) do
	{:ok, conn, protocol, nil}
	else
	error ->
	Logger.warning(
	"Opening connection to #{compose_uri_log(uri)} failed with error #{inspect(error)}"
	)

	error
	end
	end

	defp destination_opts(%URI{host: host, port: port}) do
	host = Pleroma.HTTP.AdapterHelper.parse_host(host)
	%{host: host, port: port}
	end

	defp add_http2_opts(opts, "https", tls_opts) do
	Map.merge(opts, %{protocols: [:http2, :http], transport: :tls, tls_opts: tls_opts})
	end

	defp add_http2_opts(opts, _, _), do: opts

	defp add_proxy_auth(connect_opts, %{proxy_auth: {username, password}})
	when is_binary(username) and is_binary(password) do
	Map.merge(connect_opts, %{username: username, password: password})
	end

	defp add_proxy_auth(connect_opts, _), do: connect_opts

	defp add_socks_proxy_auth(socks_opts, %{proxy_auth: {username, password}})
	when is_binary(username) and is_binary(password) do
	Map.put(socks_opts, :auth, [{:username_password, username, password}])
	end

	defp add_socks_proxy_auth(socks_opts, _), do: socks_opts

	defp proxy_open_opts(opts) do
	proxy_tls_opts = Map.get(opts, :proxy_tls_opts, [])

	opts =
	opts
	\|> Map.drop([:proxy, :proxy_auth, :proxy_tls_opts, :tls_opts])
	\|> Map.put_new(:transport, :tcp)

	if opts.transport == :tls do
	Map.put(opts, :tls_opts, proxy_tls_opts)
	else
	opts
	end
	end

	defp await_up(conn, timeout) do
	case Gun.await_up(conn, timeout) do
	{:ok, protocol} ->
	{:ok, protocol}

	error ->
	Gun.close(conn)
	error
	end
	end

	defp await_proxy_connect(stream, conn, timeout) do
	case Gun.await(conn, stream) do
	{:response, :fin, 200, _headers} ->
	case await_tunnel_up(conn, stream, timeout) do
	{:ok, protocol} -> {:ok, conn, protocol, stream}
	error -> error
	end

	- {:response, _fin, 403, _headers} ->
	- close_with_error(conn, :unauthorized)
	-
	{:response, _fin, 407, _headers} ->
	close_with_error(conn, :proxy_auth_failed)

	+ {:response, _fin, status, _headers} when is_integer(status) ->
	+ close_with_error(conn, {:proxy_connect_failed, status})
	+
	error ->
	Gun.close(conn)
	error
	end
	end

	defp close_with_error(conn, reason) do
	Gun.close(conn)
	{:error, reason}
	end

	defp transport("https"), do: :tls
	defp transport(_), do: :tcp

	defp await_tunnel_up(conn, stream_ref, timeout) do
	case Gun.await_tunnel_up(conn, stream_ref, timeout) do
	{:ok, protocol} ->
	{:ok, protocol}

	error ->
	Gun.close(conn)
	error
	end
	end

	def compose_uri_log(%URI{scheme: scheme, host: host, path: path}) do
	"#{scheme}://#{host}#{path}"
	end
	end
	diff --git a/lib/pleroma/gun/connection_pool.ex b/lib/pleroma/gun/connection_pool.ex
	index f2620836e..5e4fc5c3f 100644
	--- a/lib/pleroma/gun/connection_pool.ex
	+++ b/lib/pleroma/gun/connection_pool.ex
	@@ -1,261 +1,261 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Gun.ConnectionPool do
	@registry __MODULE__
	@locks Module.concat(__MODULE__, Locks)

	alias Pleroma.Gun.ConnectionPool.WorkerSupervisor

	@connection_option_keys [
	:http_opts,
	:http2_opts,
	:protocols,
	:proxy,
	:proxy_auth,
	:proxy_tls_opts,
	:retry,
	:retry_timeout,
	:socks_opts,
	:tcp_opts,
	:tls_opts,
	:transport,
	:ws_opts
	]

	def children do
	[
	{Registry, keys: :duplicate, name: @registry},
	{Registry, keys: :unique, name: @locks},
	Pleroma.Gun.ConnectionPool.WorkerSupervisor
	]
	end

	@spec get_conn(URI.t(), keyword()) :: {:ok, pid()} \| {:error, term()}
	def get_conn(uri, opts) do
	key = connection_key(uri, opts)

	case checkout_existing(Registry.lookup(@registry, key)) do
	{:ok, gun_pid} ->
	{:ok, gun_pid}

	:not_available ->
	provision_or_wait(key, uri, opts)
	end
	end

	defp checkout_existing([]), do: :not_available

	defp checkout_existing([{_worker_pid, nil} \| workers]), do: checkout_existing(workers)

	defp checkout_existing([{worker_pid, _value} \| workers]) do
	case checkout_worker(worker_pid) do
	{:ok, gun_pid} -> {:ok, gun_pid}
	_ -> checkout_existing(workers)
	end
	end

	defp provision_or_wait(key, uri, opts) do
	result = with_route_lock(key, fn -> provision(key, uri, opts) end)

	case result do
	{:ok, gun_pid} ->
	{:ok, gun_pid}

	{:new, worker_pid} ->
	get_gun_pid_from_worker(worker_pid)

	{:wait, worker_pid} ->
	case checkout_worker(worker_pid, :infinity) do
	{:ok, gun_pid} -> {:ok, gun_pid}
	:busy -> get_conn(uri, opts)
	error -> error
	end

	error ->
	error
	end
	end

	defp provision(key, uri, opts) do
	workers = Registry.lookup(@registry, key)

	case checkout_existing(workers) do
	{:ok, gun_pid} ->
	{:ok, gun_pid}

	:not_available ->
	case Enum.find(workers, fn {_worker_pid, value} -> is_nil(value) end) do
	{worker_pid, nil} -> {:wait, worker_pid}
	nil -> start_worker(key, uri, opts)
	end
	end
	end

	defp with_route_lock(key, function) do
	case Registry.register(@locks, key, nil) do
	{:ok, _} ->
	try do
	function.()
	after
	Registry.unregister(@locks, key)
	end

	{:error, {:already_registered, _owner}} ->
	Process.sleep(1)
	with_route_lock(key, function)
	end
	end

	defp start_worker(key, uri, opts) do
	# :gun.set_owner fails in :connected state, so the worker opens the connection itself.
	case WorkerSupervisor.start_worker([key, uri, opts, self()]) do
	{:ok, worker_pid} -> {:new, worker_pid}
	error -> error
	end
	end

	defp checkout_worker(worker_pid, timeout \\ 5_000) do
	try do
	GenServer.call(worker_pid, :checkout, timeout)
	catch
	:exit, reason -> {:error, {:connection_worker_exit, reason}}
	end
	end

	defp get_gun_pid_from_worker(worker_pid) do
	# GenServer.call will block the process for timeout length if
	# the server crashes on startup (which will happen if gun fails to connect)
	# so instead we use cast + monitor

	ref = Process.monitor(worker_pid)

	receive do
	{:conn_pid, pid} ->
	Process.demonitor(ref)
	{:ok, pid}

	{:DOWN, ^ref, :process, ^worker_pid, reason} ->
	case reason do
	{:shutdown, {:error, _} = error} -> error
	{:shutdown, error} -> {:error, error}
	_ -> {:error, reason}
	end
	end
	end

	@spec release_conn(pid()) :: :ok
	def release_conn(conn_pid) do
	# :ets.fun2ms(fn {_, {worker_pid, {gun_pid, _tunnel}}} when gun_pid == conn_pid ->
	# worker_pid end)
	query_result =
	Registry.select(@registry, [
	{{:_, :"$1", {:"$2", :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}
	])

	case query_result do
	[worker_pid] ->
	try do
	GenServer.call(worker_pid, :remove_client)
	catch
	:exit, _ -> :ok
	end

	[] ->
	:ok
	end
	end

	@spec discard_conn(pid()) :: :ok
	def discard_conn(conn_pid) do
	case worker_for_conn(conn_pid) do
	nil ->
	:ok

	worker_pid ->
	DynamicSupervisor.terminate_child(WorkerSupervisor, worker_pid)
	end
	end

	@spec cancel_stream(pid(), reference() \| [reference()]) :: :ok
	def cancel_stream(conn_pid, stream) do
	case worker_for_conn(conn_pid) do
	nil ->
	Pleroma.Gun.cancel(conn_pid, stream)

	worker_pid ->
	try do
	GenServer.call(worker_pid, {:cancel_stream, stream})
	catch
	:exit, _ -> :ok
	end
	end
	end

	@spec register_stream(pid(), reference() \| [reference()]) :: :ok
	def register_stream(conn_pid, stream) do
	call_worker(conn_pid, {:register_stream, stream})
	end

	@spec finish_stream(pid(), reference() \| [reference()]) :: :ok
	def finish_stream(conn_pid, stream) do
	call_worker(conn_pid, {:finish_stream, stream})
	end

	@spec release_stream(pid(), reference() \| [reference()]) :: :ok
	def release_stream(conn_pid, stream) do
	call_worker(conn_pid, {:release_stream, stream})
	end

	- @spec tunnel_ref(pid()) :: reference() \| nil
	+ @spec tunnel_ref(pid()) :: reference() \| :forward_proxy \| nil
	def tunnel_ref(conn_pid) do
	case Registry.select(@registry, [
	{{:_, :_, {:"$1", :"$2"}}, [{:==, :"$1", conn_pid}], [:"$2"]}
	]) do
	[tunnel] -> tunnel
	[] -> nil
	end
	end

	defp worker_for_conn(conn_pid) do
	case Registry.select(@registry, [
	{{:_, :"$1", {:"$2", :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}
	]) do
	[worker_pid] -> worker_pid
	[] -> nil
	end
	end

	defp call_worker(conn_pid, message) do
	case worker_for_conn(conn_pid) do
	nil ->
	:ok

	worker_pid ->
	try do
	GenServer.call(worker_pid, message)
	catch
	:exit, _ -> :ok
	end
	end
	end

	defp connection_key(uri, opts) do
	origin = {
	String.downcase(uri.scheme),
	String.downcase(uri.host),
	uri.port
	}

	profile =
	opts
	\|> Map.new()
	\|> Map.take(@connection_option_keys)
	\|> :erlang.term_to_binary()
	\|> then(&:crypto.hash(:sha256, &1))

	fingerprint = Base.encode16(profile, case: :lower)
	Enum.join(Tuple.to_list(origin) ++ [fingerprint], ":")
	end
	end
	diff --git a/lib/pleroma/http/adapter/gun.ex b/lib/pleroma/http/adapter/gun.ex
	index 61d297101..dea1798a1 100644
	--- a/lib/pleroma/http/adapter/gun.ex
	+++ b/lib/pleroma/http/adapter/gun.ex
	@@ -1,216 +1,253 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.HTTP.Adapter.Gun do
	@behaviour Tesla.Adapter

	alias Pleroma.Gun.ConnectionPool
	alias Tesla.Multipart

	@default_timeout 1_000

	@impl Tesla.Adapter
	def call(env, opts) do
	adapter_opts =
	Tesla.Adapter.opts(
	[close_conn: true, body_as: :plain, send_body: :at_once, receive: true],
	env,
	opts
	)
	\|> Map.new()

	with {:ok, status, headers, body} <- request(env, adapter_opts) do
	env = %{env \| status: status, headers: format_headers(headers), body: body}
	{:ok, put_stream_ref(env, body)}
	end
	end

	defp put_stream_ref(env, %{stream: stream}) do
	put_in(env.opts[:adapter][:stream], stream)
	end

	defp put_stream_ref(env, _body), do: env

	defp request(env, %{conn: conn, tunnel: tunnel} = opts) do
	uri = URI.parse(Tesla.build_url(env))
	- path = Tesla.Adapter.Shared.prepare_path(uri.path, uri.query)
	+ path = request_path(uri, tunnel)
	method = Tesla.Adapter.Shared.format_method(env.method)
	{headers, body, send_body} = prepare_body(env.headers, env.body, opts[:send_body])
	+ headers = maybe_add_forward_proxy_headers(headers, uri, opts, tunnel)
	request_opts = request_opts(opts[:reply_to], tunnel)

	stream =
	try do
	open_stream(conn, method, path, headers, body, request_opts, send_body)
	catch
	kind, reason ->
	ConnectionPool.release_conn(conn)
	:erlang.raise(kind, reason, __STACKTRACE__)
	end

	:ok = ConnectionPool.register_stream(conn, stream)
	:ok = send_stream_body(conn, stream, body, send_body)
	response = read_response(conn, stream, opts)

	case response do
	{:error, reason} ->
	ConnectionPool.cancel_stream(conn, stream)
	{:error, {:gun_stream_error, reason}}

	response ->
	unless opts[:body_as] in [:chunks, :stream] and stream_body?(response) do
	:ok = ConnectionPool.finish_stream(conn, stream)
	end

	if opts[:close_conn] and opts[:body_as] not in [:stream, :chunks] do
	:gun.close(conn)
	end

	response
	end
	end

	defp stream_body?({:ok, _status, _headers, %{stream: _stream}}), do: true
	defp stream_body?({:ok, _status, _headers, %Stream{}}), do: true
	defp stream_body?(_response), do: false

	defp request_opts(reply_to, nil), do: %{reply_to: reply_to \|\| self()}

	+ defp request_opts(reply_to, :forward_proxy), do: %{reply_to: reply_to \|\| self()}
	+
	defp request_opts(reply_to, tunnel) do
	%{reply_to: reply_to \|\| self(), tunnel: tunnel}
	end

	+ defp request_path(uri, :forward_proxy) do
	+ path = Tesla.Adapter.Shared.prepare_path(uri.path, uri.query)
	+ "#{uri.scheme}://#{authority(uri)}#{path}"
	+ end
	+
	+ defp request_path(uri, _tunnel) do
	+ Tesla.Adapter.Shared.prepare_path(uri.path, uri.query)
	+ end
	+
	+ defp maybe_add_forward_proxy_headers(headers, uri, opts, :forward_proxy) do
	+ headers
	+ \|> put_header_new("host", authority(uri))
	+ \|> maybe_put_proxy_authorization(opts)
	+ end
	+
	+ defp maybe_add_forward_proxy_headers(headers, _uri, _opts, _tunnel), do: headers
	+
	+ defp maybe_put_proxy_authorization(headers, %{proxy_auth: {username, password}})
	+ when is_binary(username) and is_binary(password) do
	+ value = "Basic " <> Base.encode64(username <> ":" <> password)
	+ List.keystore(headers, "proxy-authorization", 0, {"proxy-authorization", value})
	+ end
	+
	+ defp maybe_put_proxy_authorization(headers, _opts), do: headers
	+
	+ defp put_header_new(headers, name, value) do
	+ if List.keymember?(headers, name, 0), do: headers, else: [{name, value} \| headers]
	+ end
	+
	+ defp authority(%URI{scheme: scheme, host: host, port: port}) do
	+ host = if String.contains?(host, ":"), do: "[#{host}]", else: host
	+ if port == URI.default_port(scheme), do: host, else: "#{host}:#{port}"
	+ end
	+
	defp prepare_body(headers, %Multipart{} = multipart, _send_body) do
	{format_headers(headers ++ Multipart.headers(multipart)), Multipart.body(multipart), :stream}
	end

	defp prepare_body(headers, body, _send_body)
	when is_struct(body, Stream) or is_function(body) do
	{format_headers(headers), body, :stream}
	end

	defp prepare_body(headers, body, send_body) do
	{format_headers(headers), body \|\| "", send_body}
	end

	defp open_stream(conn, method, path, headers, _body, request_opts, :stream) do
	:gun.headers(conn, method, path, headers, request_opts)
	end

	defp open_stream(conn, method, path, headers, body, request_opts, :at_once) do
	:gun.request(conn, method, path, headers, body, request_opts)
	end

	defp send_stream_body(conn, stream, body, :stream) do
	try do
	for data <- body, do: :ok = :gun.data(conn, stream, :nofin, data)
	:gun.data(conn, stream, :fin, "")
	catch
	kind, reason ->
	ConnectionPool.cancel_stream(conn, stream)
	:erlang.raise(kind, reason, __STACKTRACE__)
	end
	end

	defp send_stream_body(_conn, _stream, _body, :at_once), do: :ok

	defp read_response(conn, stream, opts) do
	receive? = opts[:receive]

	receive do
	{:gun_response, ^conn, ^stream, :fin, status, headers} ->
	{:ok, status, headers, ""}

	{:gun_response, ^conn, ^stream, :nofin, status, headers} ->
	format_response(conn, stream, opts, status, headers, opts[:body_as])

	{:gun_up, ^conn, _protocol} when receive? ->
	read_response(conn, stream, opts)

	{:gun_error, ^conn, reason} ->
	{:error, reason}

	{:gun_error, ^conn, ^stream, reason} ->
	{:error, reason}

	{:gun_down, ^conn, _protocol, _reason, _killed_streams} when receive? ->
	read_response(conn, stream, opts)

	{:DOWN, _ref, :process, ^conn, reason} ->
	{:error, reason}
	after
	opts[:timeout] \|\| @default_timeout -> {:error, :recv_response_timeout}
	end
	end

	defp format_response(conn, stream, opts, status, headers, :plain) do
	case read_body(conn, stream, opts) do
	{:ok, body} ->
	{:ok, status, headers, body}

	{:error, error} ->
	:ok = :gun.flush(stream)
	{:error, error}
	end
	end

	defp format_response(conn, stream, opts, status, headers, :stream) do
	body =
	Stream.resource(
	fn -> :reading end,
	fn
	:reading ->
	case Tesla.Adapter.Gun.read_chunk(conn, stream, opts) do
	{:nofin, part} -> {[part], :reading}
	{:fin, part} -> {[part], :done}
	{:error, reason} -> raise "Gun stream failed: #{inspect(reason)}"
	end

	:done ->
	{:halt, :done}
	end,
	fn _state ->
	if opts[:close_conn], do: :gun.close(conn)
	end
	)

	{:ok, status, headers, body}
	end

	defp format_response(conn, stream, opts, status, headers, :chunks) do
	{:ok, status, headers, %{pid: conn, stream: stream, opts: Map.to_list(opts)}}
	end

	defp read_body(conn, stream, opts, acc \\ "") do
	receive do
	{:gun_data, ^conn, ^stream, :fin, body} ->
	append_body(acc, body, opts[:max_body])

	{:gun_data, ^conn, ^stream, :nofin, body} ->
	with {:ok, acc} <- append_body(acc, body, opts[:max_body]) do
	read_body(conn, stream, opts, acc)
	end

	{:gun_error, ^conn, ^stream, reason} ->
	{:error, reason}

	{:DOWN, _ref, :process, ^conn, reason} ->
	{:error, reason}
	after
	opts[:timeout] \|\| @default_timeout -> {:error, :recv_body_timeout}
	end
	end

	defp append_body(acc, part, nil), do: {:ok, acc <> part}

	defp append_body(acc, part, limit) do
	body = acc <> part
	if byte_size(body) <= limit, do: {:ok, body}, else: {:error, :body_too_large}
	end

	defp format_headers(headers) do
	Enum.map(headers, fn {key, value} ->
	{String.downcase(to_string(key)), to_string(value)}
	end)
	end
	end
	diff --git a/test/pleroma/gun/conn_test.exs b/test/pleroma/gun/conn_test.exs
	index 79ea7b33f..a38740a42 100644
	--- a/test/pleroma/gun/conn_test.exs
	+++ b/test/pleroma/gun/conn_test.exs
	@@ -1,144 +1,179 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Gun.ConnTest do
	use ExUnit.Case

	import Mox

	alias Pleroma.Gun.Conn

	setup :verify_on_exit!

	setup do
	stub(Pleroma.GunMock, :await_up, fn _, _ -> {:ok, :http} end)
	stub(Pleroma.GunMock, :await_tunnel_up, fn _, _, _ -> {:ok, :http2} end)
	stub(Pleroma.GunMock, :close, fn _ -> :ok end)

	:ok
	end

	test "selects the direct transport from the URI scheme" do
	expect(Pleroma.GunMock, :open, 2, fn
	~c"example.com", 8443, %{transport: :tls, tls_opts: tls_opts} ->
	assert tls_opts[:verify] == :verify_peer
	{:ok, spawn_link(fn -> Process.sleep(:infinity) end)}

	~c"example.com", 443, %{transport: :tcp} ->
	{:ok, spawn_link(fn -> Process.sleep(:infinity) end)}
	end)

	assert {:ok, https_conn, :http, nil} = Conn.open(URI.parse("https://example.com:8443"), [])
	assert {:ok, http_conn, :http, nil} = Conn.open(URI.parse("http://example.com:443"), [])

	stop_conn(https_conn)
	stop_conn(http_conn)
	end

	test "passes authentication to an HTTP CONNECT proxy" do
	conn = spawn_link(fn -> Process.sleep(:infinity) end)
	stream = make_ref()

	expect(Pleroma.GunMock, :open, fn ~c"proxy.example", 8080, opts ->
	assert opts.transport == :tcp
	refute Map.has_key?(opts, :tls_opts)
	{:ok, conn}
	end)

	expect(Pleroma.GunMock, :connect, fn ^conn, connect_opts ->
	assert connect_opts.username == "alice"
	assert connect_opts.password == "secret"
	assert connect_opts.transport == :tls
	assert connect_opts.protocols == [:http2, :http]
	stream
	end)

	expect(Pleroma.GunMock, :await, fn ^conn, ^stream -> {:response, :fin, 200, []} end)

	assert {:ok, ^conn, :http2, ^stream} =
	Conn.open(URI.parse("https://origin.example/inbox"),
	proxy: {~c"proxy.example", 8080},
	proxy_auth: {"alice", "secret"}
	)

	stop_conn(conn)
	end

	test "uses separate TLS options for a TLS CONNECT proxy" do
	conn = spawn_link(fn -> Process.sleep(:infinity) end)
	stream = make_ref()

	expect(Pleroma.GunMock, :open, fn ~c"proxy.example", 8443, opts ->
	assert opts.transport == :tls
	assert opts.tls_opts == [verify: :verify_none]
	{:ok, conn}
	end)

	expect(Pleroma.GunMock, :connect, fn ^conn, _connect_opts -> stream end)
	expect(Pleroma.GunMock, :await, fn ^conn, ^stream -> {:response, :fin, 200, []} end)

	assert {:ok, ^conn, :http2, ^stream} =
	Conn.open(URI.parse("https://origin.example/inbox"),
	proxy: {~c"proxy.example", 8443},
	transport: :tls,
	proxy_tls_opts: [verify: :verify_none]
	)

	stop_conn(conn)
	end

	+ test "uses HTTP/1 for a TLS forwarding proxy" do
	+ conn = spawn_link(fn -> Process.sleep(:infinity) end)
	+
	+ expect(Pleroma.GunMock, :open, fn ~c"proxy.example", 8443, opts ->
	+ assert opts.transport == :tls
	+ assert opts.tls_opts == [verify: :verify_none]
	+ assert opts.protocols == [:http]
	+ {:ok, conn}
	+ end)
	+
	+ assert {:ok, ^conn, :http, :forward_proxy} =
	+ Conn.open(URI.parse("http://origin.example/inbox"),
	+ proxy: {~c"proxy.example", 8443},
	+ transport: :tls,
	+ proxy_tls_opts: [verify: :verify_none]
	+ )
	+
	+ stop_conn(conn)
	+ end
	+
	test "passes authentication and remote DNS destination to a SOCKS5 proxy" do
	conn = spawn_link(fn -> Process.sleep(:infinity) end)

	expect(Pleroma.GunMock, :open, fn ~c"proxy.example", 1080, opts ->
	assert opts.protocols == [:socks]
	assert opts.transport == :tcp
	assert opts.socks_opts.host == ~c"origin.example"
	assert opts.socks_opts.port == 443
	assert opts.socks_opts.auth == [{:username_password, "alice", "secret"}]
	assert opts.socks_opts.transport == :tls
	assert opts.socks_opts.protocols == [:http2, :http]
	refute Map.has_key?(opts, :tls_opts)
	{:ok, conn}
	end)

	stub(Pleroma.GunMock, :await_up, fn _, _ -> {:ok, :socks} end)

	assert {:ok, ^conn, :http2, nil} =
	Conn.open(URI.parse("https://origin.example/inbox"),
	proxy: {:socks5, ~c"proxy.example", 1080},
	proxy_auth: {"alice", "secret"}
	)

	stop_conn(conn)
	end

	test "closes a CONNECT socket when proxy authentication fails" do
	conn = spawn_link(fn -> Process.sleep(:infinity) end)
	stream = make_ref()

	expect(Pleroma.GunMock, :open, fn _, _, _ -> {:ok, conn} end)
	expect(Pleroma.GunMock, :connect, fn ^conn, _ -> stream end)
	expect(Pleroma.GunMock, :await, fn ^conn, ^stream -> {:response, :nofin, 407, []} end)
	expect(Pleroma.GunMock, :close, fn ^conn -> Process.exit(conn, :normal) end)

	assert {:error, :proxy_auth_failed} =
	Conn.open(URI.parse("https://origin.example/inbox"),
	proxy: {~c"proxy.example", 8080},
	proxy_auth: {"alice", "wrong"}
	)
	end

	+ test "returns the proxy status when CONNECT is forbidden" do
	+ conn = spawn_link(fn -> Process.sleep(:infinity) end)
	+ stream = make_ref()
	+
	+ expect(Pleroma.GunMock, :open, fn _, _, _ -> {:ok, conn} end)
	+ expect(Pleroma.GunMock, :connect, fn ^conn, _ -> stream end)
	+ expect(Pleroma.GunMock, :await, fn ^conn, ^stream -> {:response, :fin, 403, []} end)
	+ expect(Pleroma.GunMock, :close, fn ^conn -> Process.exit(conn, :normal) end)
	+
	+ assert {:error, {:proxy_connect_failed, 403}} =
	+ Conn.open(URI.parse("https://origin.example/inbox"),
	+ proxy: {~c"proxy.example", 8080}
	+ )
	+ end
	+
	test "rejects SOCKS4 before opening a socket" do
	assert {:error, :socks4_unsupported} =
	Conn.open(URI.parse("https://origin.example/inbox"),
	proxy: {:socks4, ~c"proxy.example", 1080}
	)
	end

	defp stop_conn(conn) do
	Process.unlink(conn)
	Process.exit(conn, :kill)
	end
	end
	diff --git a/test/pleroma/http/gun_integration_test.exs b/test/pleroma/http/gun_integration_test.exs
	index f50f8105e..9282e2e6c 100644
	--- a/test/pleroma/http/gun_integration_test.exs
	+++ b/test/pleroma/http/gun_integration_test.exs
	@@ -1,707 +1,808 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.HTTP.GunIntegrationTest do
	use ExUnit.Case, async: false
	use Pleroma.Tests.Helpers

	alias Pleroma.Tesla.Middleware.ConnectionPool

	setup do
	clear_config([Pleroma.Gun], Pleroma.Gun.API)
	clear_config([:connections_pool, :max_idle_time], 1_000)

	origin = start_tls_origin(self())
	connect_proxy = start_connect_proxy(self())
	+ forward_proxy = start_forward_proxy(self())
	socks_proxy = start_socks5_proxy(self())

	on_exit(fn ->
	Process.sleep(20)
	stop_server(socks_proxy)
	+ stop_server(forward_proxy)
	stop_server(connect_proxy)
	stop_server(origin)
	end)

	- {:ok, origin: origin, connect_proxy: connect_proxy, socks_proxy: socks_proxy}
	+ {:ok,
	+ origin: origin,
	+ connect_proxy: connect_proxy,
	+ forward_proxy: forward_proxy,
	+ socks_proxy: socks_proxy}
	end

	test "requests HTTPS directly on a non-standard port", %{origin: origin} do
	assert {:ok, %Tesla.Env{status: 200, body: "ok"}} = request(origin.base_url <> "/final")
	end

	test "follows redirects through an authenticated CONNECT proxy", %{
	origin: origin,
	connect_proxy: proxy
	} do
	proxy_opts = [
	proxy: {~c"127.0.0.1", proxy.port},
	proxy_auth: {"alice", "secret"}
	]

	assert {:ok, %Tesla.Env{status: 200, body: "ok"}} =
	request(origin.base_url <> "/redirect", proxy_opts, true)

	expected = "Basic " <> Base.encode64("alice:secret")
	assert_receive {:connect_proxy, ^expected, "127.0.0.1", origin_port}
	assert origin_port == origin.port
	end

	test "requests through an authenticated CONNECT proxy", %{
	origin: origin,
	connect_proxy: proxy
	} do
	proxy_opts = [
	proxy: {~c"127.0.0.1", proxy.port},
	proxy_auth: {"alice", "secret"}
	]

	result = request(origin.base_url <> "/final", proxy_opts)

	expected = "Basic " <> Base.encode64("alice:secret")
	assert_receive {:connect_proxy, ^expected, "127.0.0.1", origin_port}
	assert origin_port == origin.port
	assert worker_protocol(origin.port) == :http
	assert_receive {:origin_request, "/final"}
	assert {:ok, %Tesla.Env{status: 200, body: "ok"}} = result
	end

	+ test "requests HTTP through a forwarding proxy", %{forward_proxy: proxy} do
	+ proxy_opts = [proxy: {~c"127.0.0.1", proxy.port}]
	+ url = "http://origin.example:8080/final?key=value"
	+
	+ assert {:ok, %Tesla.Env{status: 200, body: "ok"}} = request(url, proxy_opts)
	+
	+ assert_receive {:forward_proxy, ^url, "origin.example:8080", nil}
	+ end
	+
	+ test "authenticates HTTP requests to a forwarding proxy", %{forward_proxy: proxy} do
	+ proxy_opts = [
	+ proxy: {~c"127.0.0.1", proxy.port},
	+ proxy_auth: {"alice", "secret"}
	+ ]
	+
	+ url = "http://origin.example/final"
	+ assert {:ok, %Tesla.Env{status: 200, body: "ok"}} = request(url, proxy_opts)
	+
	+ expected = "Basic " <> Base.encode64("alice:secret")
	+ assert_receive {:forward_proxy, ^url, "origin.example", ^expected}
	+ end
	+
	test "opens a request-ready authenticated CONNECT tunnel", %{
	origin: origin,
	connect_proxy: proxy
	} do
	uri = URI.parse(origin.base_url <> "/final")

	opts = [
	proxy: {~c"127.0.0.1", proxy.port},
	proxy_auth: {"alice", "secret"},
	tls_opts: [verify: :verify_none],
	connect_timeout: 2_000
	]

	assert {:ok, conn, :http, tunnel} = Pleroma.Gun.Conn.open(uri, opts)
	stream = :gun.get(conn, "/final", [], %{tunnel: tunnel})
	assert {:response, :nofin, 200, _headers} = :gun.await(conn, stream, 2_000)
	assert {:ok, "ok"} = :gun.await_body(conn, stream, 2_000)
	:gun.close(conn)
	end

	test "streams chunks through an authenticated CONNECT tunnel", %{
	origin: origin,
	connect_proxy: proxy
	} do
	proxy_opts = [
	proxy: {~c"127.0.0.1", proxy.port},
	proxy_auth: {"alice", "secret"}
	]

	assert {:ok, %Tesla.Env{status: 200, body: client}} =
	request_chunks(origin.base_url <> "/chunks", proxy_opts)

	assert collect_chunks(client) == "hello"
	assert :ok = Pleroma.Gun.ConnectionPool.release_conn(client.pid)
	end

	test "discards an unfinished HTTP/1 redirect stream before following it", %{
	origin: origin,
	connect_proxy: proxy
	} do
	proxy_opts = [
	proxy: {~c"127.0.0.1", proxy.port},
	proxy_auth: {"alice", "secret"}
	]

	assert {:ok, %Tesla.Env{status: 200, body: client}} =
	request_chunks(origin.base_url <> "/redirect_chunks", proxy_opts, true)

	assert collect_chunks(client) == "hello"
	assert :ok = Pleroma.Gun.ConnectionPool.release_conn(client.pid)

	expected = "Basic " <> Base.encode64("alice:secret")
	assert_receive {:connect_proxy, ^expected, "127.0.0.1", _port}
	assert_receive {:connect_proxy, ^expected, "127.0.0.1", _port}
	end

	test "uploads multipart bodies through an authenticated CONNECT tunnel", %{
	origin: origin,
	connect_proxy: proxy
	} do
	proxy_opts = [
	proxy: {~c"127.0.0.1", proxy.port},
	proxy_auth: {"alice", "secret"}
	]

	multipart = Tesla.Multipart.new() \|> Tesla.Multipart.add_field("name", "value")

	assert {:ok, %Tesla.Env{status: 200}} =
	request_multipart(origin.base_url <> "/upload", multipart, proxy_opts)

	assert_receive {:origin_headers, "/upload", headers}
	assert String.downcase(headers) =~ "content-type: multipart/form-data;"
	assert_receive {:origin_body, "/upload", body}
	assert body =~ "name=\"name\""
	assert body =~ "value"
	end

	test "uploads streamed bodies through an authenticated CONNECT tunnel", %{
	origin: origin,
	connect_proxy: proxy
	} do
	proxy_opts = [
	proxy: {~c"127.0.0.1", proxy.port},
	proxy_auth: {"alice", "secret"}
	]

	body = Stream.map(["streamed", "-body"], & &1)

	assert {:ok, %Tesla.Env{status: 200}} =
	request_stream(origin.base_url <> "/upload", body, proxy_opts)

	assert_receive {:origin_body, "/upload", "streamed-body"}
	end

	test "discards a partial HTTP/1 upload when body enumeration raises", %{origin: origin} do
	body =
	Stream.map(["partial", "raise"], fn
	"raise" -> raise "upload failed"
	part -> part
	end)

	assert_raise RuntimeError, "upload failed", fn ->
	request_stream(origin.base_url <> "/upload", body, [])
	end

	assert_receive {:origin_headers, "/upload", _headers}
	assert worker_protocol(origin.port) == nil

	assert {:ok, %Tesla.Env{status: 200, body: "ok"}} = request(origin.base_url <> "/final")
	end

	test "releases the lease when Gun rejects an invalid request header", %{origin: origin} do
	adapter_opts = [
	tls_opts: [verify: :verify_none],
	connect_timeout: 2_000,
	timeout: 2_000
	]

	client = Tesla.client([ConnectionPool], Tesla.Adapter.Gun)

	assert {:invalid_request_header, "x-test", _message} =
	catch_error(
	Tesla.get(client, origin.base_url <> "/final",
	headers: [{"x-test", "value\r\nx-injected: true"}],
	opts: [adapter: adapter_opts]
	)
	)

	assert worker_clients(origin.port) == %{}
	refute_receive {:origin_request, "/final"}
	assert {:ok, %Tesla.Env{status: 200, body: "ok"}} = request(origin.base_url <> "/final")
	end

	test "requests through authenticated SOCKS5 with proxy-side DNS", %{
	origin: origin,
	socks_proxy: proxy
	} do
	proxy_opts = [
	proxy: {:socks5, ~c"127.0.0.1", proxy.port},
	proxy_auth: {"alice", "secret"}
	]

	url = "https://localhost:#{origin.port}/final"
	assert {:ok, %Tesla.Env{status: 200, body: "ok"}} = request(url, proxy_opts)

	assert_receive {:socks5_proxy, "alice", "secret", "localhost", origin_port}
	assert origin_port == origin.port
	end

	+ test "requests through unauthenticated SOCKS5 with an IP proxy host", %{
	+ origin: origin,
	+ socks_proxy: proxy
	+ } do
	+ proxy_opts = [proxy: {:socks5, {127, 0, 0, 1}, proxy.port}]
	+
	+ assert {:ok, %Tesla.Env{status: 200, body: "ok"}} =
	+ request(origin.base_url <> "/final", proxy_opts)
	+
	+ assert_receive {:socks5_proxy, nil, nil, "127.0.0.1", origin_port}
	+ assert origin_port == origin.port
	+ end
	+
	+ test "requests through unauthenticated SOCKS5 with a localhost proxy host", %{
	+ origin: origin,
	+ socks_proxy: proxy
	+ } do
	+ proxy_opts = [proxy: {:socks5, ~c"localhost", proxy.port}]
	+
	+ assert {:ok, %Tesla.Env{status: 200, body: "ok"}} =
	+ request(origin.base_url <> "/final", proxy_opts)
	+
	+ assert_receive {:socks5_proxy, nil, nil, "127.0.0.1", origin_port}
	+ assert origin_port == origin.port
	+ end
	+
	defp request(url, proxy_opts \\ [], follow_redirects \\ false) do
	middleware =
	if follow_redirects do
	[Tesla.Middleware.FollowRedirects, ConnectionPool]
	else
	[ConnectionPool]
	end

	adapter_opts =
	Keyword.merge(
	[tls_opts: [verify: :verify_none], connect_timeout: 2_000, timeout: 2_000],
	proxy_opts
	)

	middleware
	\|> Tesla.client(Tesla.Adapter.Gun)
	\|> Tesla.get(url, opts: [adapter: adapter_opts])
	end

	defp request_chunks(url, proxy_opts, follow_redirects \\ false) do
	adapter_opts =
	Keyword.merge(
	[
	body_as: :chunks,
	tls_opts: [verify: :verify_none],
	connect_timeout: 2_000,
	timeout: 2_000
	],
	proxy_opts
	)

	middleware =
	if follow_redirects do
	[Tesla.Middleware.FollowRedirects, ConnectionPool]
	else
	[ConnectionPool]
	end

	middleware
	\|> Tesla.client(Tesla.Adapter.Gun)
	\|> Tesla.get(url, opts: [adapter: adapter_opts])
	end

	defp request_multipart(url, multipart, proxy_opts) do
	adapter_opts =
	Keyword.merge(
	[tls_opts: [verify: :verify_none], connect_timeout: 2_000, timeout: 2_000],
	proxy_opts
	)

	[ConnectionPool]
	\|> Tesla.client(Tesla.Adapter.Gun)
	\|> Tesla.post(url, multipart, opts: [adapter: adapter_opts])
	end

	defp request_stream(url, body, proxy_opts) do
	adapter_opts =
	Keyword.merge(
	[tls_opts: [verify: :verify_none], connect_timeout: 2_000, timeout: 2_000],
	proxy_opts
	)

	[ConnectionPool]
	\|> Tesla.client(Tesla.Adapter.Gun)
	\|> Tesla.post(url, body, opts: [adapter: adapter_opts])
	end

	defp collect_chunks(client, acc \\ "") do
	case Tesla.Adapter.Gun.read_chunk(client.pid, client.stream, client.opts) do
	{:nofin, chunk} -> collect_chunks(client, acc <> chunk)
	{:fin, chunk} -> acc <> chunk
	end
	end

	defp start_tls_origin(parent) do
	certfile = Path.expand("../../fixtures/server.pem", __DIR__)
	keyfile = Path.expand("../../fixtures/private_key.pem", __DIR__)

	{:ok, listener} =
	:ssl.listen(0, [
	:binary,
	certfile: certfile,
	keyfile: keyfile,
	reuseaddr: true,
	active: false,
	packet: :raw,
	ip: {127, 0, 0, 1}
	])

	{:ok, {{127, 0, 0, 1}, port}} = :ssl.sockname(listener)
	{:ok, acceptor} = Task.start_link(fn -> accept_tls_loop(listener, parent) end)

	%{
	listener: listener,
	acceptor: acceptor,
	transport: :ssl,
	port: port,
	base_url: "https://127.0.0.1:#{port}"
	}
	end

	defp accept_tls_loop(listener, parent) do
	case :ssl.transport_accept(listener) do
	{:ok, tcp_socket} ->
	case :ssl.handshake(tcp_socket, 2_000) do
	{:ok, socket} ->
	pid = spawn(fn -> receive_and_serve_tls(socket, parent) end)
	:ok = :ssl.controlling_process(socket, pid)
	send(pid, :serve)

	{:error, _reason} ->
	:gen_tcp.close(tcp_socket)
	end

	accept_tls_loop(listener, parent)

	{:error, _reason} ->
	:ok
	end
	end

	defp receive_and_serve_tls(socket, parent) do
	receive do
	:serve -> serve_tls(socket, parent)
	end
	end

	defp serve_tls(socket, parent) do
	case recv_headers(:ssl, socket) do
	{:ok, request} ->
	{headers, buffered_body} = split_headers(request)
	path = request_path(headers)
	send(parent, {:origin_request, path})
	send(parent, {:origin_headers, path, headers})

	case path do
	"/redirect" ->
	send_tls_response(socket, 302, "Found", [{"Location", "/final"}], "")

	"/redirect_chunks" ->
	send_tls_response(socket, 302, "Found", [{"Location", "/chunks"}], "discard-me")

	"/final" ->
	send_tls_response(socket, 200, "OK", [], "ok")

	"/chunks" ->
	send_tls_chunks(socket, ["hel", "lo"])

	"/upload" ->
	{:ok, body} = recv_request_body(socket, headers, buffered_body)
	send(parent, {:origin_body, path, body})
	send_tls_response(socket, 200, "OK", [], "uploaded")

	_ ->
	send_tls_response(socket, 404, "Not Found", [], "not found")
	end

	serve_tls(socket, parent)

	{:error, _reason} ->
	:ssl.close(socket)
	end
	end

	defp send_tls_response(socket, status, reason, headers, body) do
	headers =
	[
	{"Content-Length", Integer.to_string(byte_size(body))},
	{"Connection", "keep-alive"}
	] ++ headers

	:ssl.send(socket, [
	"HTTP/1.1 ",
	Integer.to_string(status),
	" ",
	reason,
	"\r\n",
	Enum.map(headers, fn {key, value} -> [key, ": ", value, "\r\n"] end),
	"\r\n",
	body
	])
	end

	defp send_tls_chunks(socket, chunks) do
	:ssl.send(socket, [
	"HTTP/1.1 200 OK\r\n",
	"Transfer-Encoding: chunked\r\n",
	"Connection: keep-alive\r\n\r\n",
	Enum.map(chunks, fn chunk ->
	[Integer.to_string(byte_size(chunk), 16), "\r\n", chunk, "\r\n"]
	end),
	"0\r\n\r\n"
	])
	end

	defp start_connect_proxy(parent) do
	start_tcp_server(fn socket -> serve_connect_proxy(socket, parent) end)
	end

	defp serve_connect_proxy(socket, parent) do
	with {:ok, headers} <- recv_headers(:gen_tcp, socket),
	{:ok, host, port} <- parse_connect(headers),
	auth when is_binary(auth) <- header(headers, "proxy-authorization"),
	{:ok, upstream} <- tcp_connect(host, port) do
	send(parent, {:connect_proxy, auth, host, port})
	:ok = :gen_tcp.send(socket, "HTTP/1.1 200 Connection established\r\n\r\n")
	tunnel(socket, upstream)
	else
	_ ->
	:gen_tcp.send(
	socket,
	"HTTP/1.1 407 Proxy Authentication Required\r\nContent-Length: 0\r\n\r\n"
	)

	:gen_tcp.close(socket)
	end
	end

	+ defp start_forward_proxy(parent) do
	+ start_tcp_server(fn socket -> serve_forward_proxy(socket, parent) end)
	+ end
	+
	+ defp serve_forward_proxy(socket, parent) do
	+ with {:ok, request} <- recv_headers(:gen_tcp, socket) do
	+ {headers, _buffered_body} = split_headers(request)
	+ target = request_path(headers)
	+ host = header(headers, "host")
	+ auth = header(headers, "proxy-authorization")
	+
	+ if String.starts_with?(target, "http://") do
	+ send(parent, {:forward_proxy, target, host, auth})
	+
	+ :gen_tcp.send(
	+ socket,
	+ "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: keep-alive\r\n\r\nok"
	+ )
	+
	+ serve_forward_proxy(socket, parent)
	+ else
	+ :gen_tcp.send(
	+ socket,
	+ "HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
	+ )
	+
	+ :gen_tcp.close(socket)
	+ end
	+ end
	+ end
	+
	defp start_socks5_proxy(parent) do
	start_tcp_server(fn socket -> serve_socks5_proxy(socket, parent) end)
	end

	defp serve_socks5_proxy(socket, parent) do
	with {:ok, <<5, method_count>>} <- :gen_tcp.recv(socket, 2, 2_000),
	{:ok, methods} <- :gen_tcp.recv(socket, method_count, 2_000),
	- true <- :binary.match(methods, <<2>>) != :nomatch,
	- :ok <- :gen_tcp.send(socket, <<5, 2>>),
	- {:ok, username, password} <- recv_socks5_auth(socket),
	- :ok <- :gen_tcp.send(socket, <<1, 0>>),
	+ {:ok, username, password} <- negotiate_socks5_auth(socket, methods),
	{:ok, host, port} <- recv_socks5_destination(socket),
	{:ok, upstream} <- tcp_connect(host, port),
	:ok <- :gen_tcp.send(socket, <<5, 0, 0, 1, 0, 0, 0, 0, 0, 0>>) do
	send(parent, {:socks5_proxy, username, password, host, port})
	tunnel(socket, upstream)
	else
	_ ->
	:gen_tcp.close(socket)
	end
	end

	+ defp negotiate_socks5_auth(socket, methods) do
	+ cond do
	+ :binary.match(methods, <<2>>) != :nomatch ->
	+ with :ok <- :gen_tcp.send(socket, <<5, 2>>),
	+ {:ok, username, password} <- recv_socks5_auth(socket),
	+ :ok <- :gen_tcp.send(socket, <<1, 0>>) do
	+ {:ok, username, password}
	+ end
	+
	+ :binary.match(methods, <<0>>) != :nomatch ->
	+ with :ok <- :gen_tcp.send(socket, <<5, 0>>) do
	+ {:ok, nil, nil}
	+ end
	+
	+ true ->
	+ {:error, :unsupported_auth}
	+ end
	+ end
	+
	defp recv_socks5_auth(socket) do
	with {:ok, <<1, username_length>>} <- :gen_tcp.recv(socket, 2, 2_000),
	{:ok, username} <- :gen_tcp.recv(socket, username_length, 2_000),
	{:ok, <<password_length>>} <- :gen_tcp.recv(socket, 1, 2_000),
	{:ok, password} <- :gen_tcp.recv(socket, password_length, 2_000) do
	{:ok, username, password}
	end
	end

	defp recv_socks5_destination(socket) do
	with {:ok, <<5, 1, 0, address_type>>} <- :gen_tcp.recv(socket, 4, 2_000),
	{:ok, host} <- recv_socks5_host(socket, address_type),
	{:ok, <<port::16>>} <- :gen_tcp.recv(socket, 2, 2_000) do
	{:ok, host, port}
	end
	end

	defp recv_socks5_host(socket, 1) do
	with {:ok, <<a, b, c, d>>} <- :gen_tcp.recv(socket, 4, 2_000) do
	{:ok, Enum.join([a, b, c, d], ".")}
	end
	end

	defp recv_socks5_host(socket, 3) do
	with {:ok, <<length>>} <- :gen_tcp.recv(socket, 1, 2_000),
	{:ok, host} <- :gen_tcp.recv(socket, length, 2_000) do
	{:ok, host}
	end
	end

	defp start_tcp_server(handler) do
	{:ok, listener} =
	:gen_tcp.listen(0, [
	:binary,
	active: false,
	packet: :raw,
	reuseaddr: true,
	ip: {127, 0, 0, 1}
	])

	{:ok, {{127, 0, 0, 1}, port}} = :inet.sockname(listener)
	{:ok, acceptor} = Task.start_link(fn -> accept_tcp_loop(listener, handler) end)
	%{listener: listener, acceptor: acceptor, transport: :gen_tcp, port: port}
	end

	defp accept_tcp_loop(listener, handler) do
	case :gen_tcp.accept(listener) do
	{:ok, socket} ->
	pid = spawn(fn -> receive_and_serve_tcp(socket, handler) end)
	:ok = :gen_tcp.controlling_process(socket, pid)
	send(pid, :serve)
	accept_tcp_loop(listener, handler)

	{:error, _reason} ->
	:ok
	end
	end

	defp receive_and_serve_tcp(socket, handler) do
	receive do
	:serve -> handler.(socket)
	end
	end

	defp recv_headers(transport, socket, acc \\ <<>>) do
	case transport.recv(socket, 0, 2_000) do
	{:ok, data} ->
	acc = acc <> data

	if :binary.match(acc, "\r\n\r\n") == :nomatch do
	recv_headers(transport, socket, acc)
	else
	{:ok, acc}
	end

	{:error, _reason} = error ->
	error
	end
	end

	defp request_path(headers) do
	headers
	\|> String.split("\r\n", parts: 2)
	\|> hd()
	\|> String.split(" ")
	\|> Enum.at(1)
	end

	defp split_headers(request) do
	{index, 4} = :binary.match(request, "\r\n\r\n")
	split_at = index + 4
	<<headers::binary-size(split_at), body::binary>> = request
	{headers, body}
	end

	defp recv_request_body(socket, headers, buffered_body) do
	cond do
	String.contains?(String.downcase(headers), "transfer-encoding: chunked") ->
	recv_chunked_body(socket, buffered_body, "")

	content_length = header(headers, "content-length") ->
	recv_exact(socket, buffered_body, String.to_integer(content_length))

	true ->
	{:ok, buffered_body}
	end
	end

	defp recv_chunked_body(socket, buffer, acc) do
	with {:ok, line, buffer} <- recv_line(socket, buffer),
	{size, ""} <- Integer.parse(line, 16) do
	if size == 0 do
	{:ok, acc}
	else
	with {:ok, chunk_and_crlf, buffer} <- take_bytes(socket, buffer, size + 2),
	<<chunk::binary-size(size), "\r\n">> <- chunk_and_crlf do
	recv_chunked_body(socket, buffer, acc <> chunk)
	end
	end
	end
	end

	defp recv_line(socket, buffer) do
	case :binary.match(buffer, "\r\n") do
	{index, 2} ->
	<<line::binary-size(index), "\r\n", rest::binary>> = buffer
	{:ok, line, rest}

	:nomatch ->
	with {:ok, data} <- :ssl.recv(socket, 0, 2_000) do
	recv_line(socket, buffer <> data)
	end
	end
	end

	defp take_bytes(_socket, buffer, size) when byte_size(buffer) >= size do
	<<bytes::binary-size(size), rest::binary>> = buffer
	{:ok, bytes, rest}
	end

	defp take_bytes(socket, buffer, size) do
	with {:ok, data} <- :ssl.recv(socket, 0, 2_000) do
	take_bytes(socket, buffer <> data, size)
	end
	end

	defp recv_exact(socket, buffer, size) do
	with {:ok, body, _rest} <- take_bytes(socket, buffer, size) do
	{:ok, body}
	end
	end

	defp parse_connect(headers) do
	case headers \|> String.split("\r\n", parts: 2) \|> hd() \|> String.split(" ") do
	["CONNECT", authority, _protocol] ->
	case String.split(authority, ":", parts: 2) do
	[host, port] -> {:ok, host, String.to_integer(port)}
	_ -> {:error, :invalid_authority}
	end

	_ ->
	{:error, :invalid_connect}
	end
	end

	defp header(headers, wanted_name) do
	headers
	\|> String.split("\r\n")
	\|> Enum.find_value(fn line ->
	case String.split(line, ":", parts: 2) do
	[name, value] ->
	if String.downcase(name) == wanted_name, do: String.trim(value)

	_ ->
	nil
	end
	end)
	end

	defp tcp_connect(host, port) do
	:gen_tcp.connect(
	String.to_charlist(host),
	port,
	[:binary, active: false, packet: :raw],
	2_000
	)
	end

	defp worker_protocol(port) do
	case worker_state(port) do
	nil -> nil
	state -> state.protocol
	end
	end

	defp worker_clients(port) do
	case worker_state(port) do
	nil -> nil
	state -> state.clients
	end
	end

	defp worker_state(port) do
	prefix = "https:127.0.0.1:#{port}:"

	Pleroma.Gun.ConnectionPool
	\|> Registry.select([{{:"$1", :"$2", :_}, [], [{{:"$1", :"$2"}}]}])
	\|> Enum.find_value(fn {key, worker} ->
	if String.starts_with?(key, prefix), do: :sys.get_state(worker)
	end)
	end

	defp tunnel(client, upstream) do
	:ok = :inet.setopts(client, active: :once)
	:ok = :inet.setopts(upstream, active: :once)
	tunnel_loop(client, upstream)
	end

	defp tunnel_loop(client, upstream) do
	receive do
	{:tcp, ^client, data} ->
	:ok = :gen_tcp.send(upstream, data)
	:ok = :inet.setopts(client, active: :once)
	tunnel_loop(client, upstream)

	{:tcp, ^upstream, data} ->
	:ok = :gen_tcp.send(client, data)
	:ok = :inet.setopts(upstream, active: :once)
	tunnel_loop(client, upstream)

	{:tcp_closed, _socket} ->
	:gen_tcp.close(client)
	:gen_tcp.close(upstream)

	{:tcp_error, _socket, _reason} ->
	:gen_tcp.close(client)
	:gen_tcp.close(upstream)
	after
	10_000 ->
	:gen_tcp.close(client)
	:gen_tcp.close(upstream)
	end
	end

	defp stop_server(%{listener: listener, acceptor: acceptor, transport: transport}) do
	transport.close(listener)

	if Process.alive?(acceptor), do: Process.exit(acceptor, :normal)
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Fri, Oct 30, 7:33 PM (14 h, 28 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1800471
Default Alt Text: (55 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions