No OneTemporary
Actions

Size

27 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
	index 4243f115d..e3b94c5cc 100644
	--- a/lib/pleroma/web/activity_pub/publisher.ex
	+++ b/lib/pleroma/web/activity_pub/publisher.ex
	@@ -1,297 +1,298 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.Publisher do
	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.Delivery
	alias Pleroma.HTTP
	alias Pleroma.Instances
	alias Pleroma.Object
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Relay
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.FedSockets

	require Pleroma.Constants

	import Pleroma.Web.ActivityPub.Visibility

	@behaviour Pleroma.Web.Federator.Publisher

	require Logger

	@moduledoc """
	ActivityPub outgoing federation module.
	"""

	@doc """
	Determine if an activity can be represented by running it through Transmogrifier.
	"""
	def is_representable?(%Activity{} = activity) do
	with {:ok, _data} <- Transmogrifier.prepare_outgoing(activity.data) do
	true
	else
	_e ->
	false
	end
	end

	@doc """
	Publish a single message to a peer. Takes a struct with the following
	parameters set:

	* `inbox`: the inbox to publish to
	* `json`: the JSON message body representing the ActivityPub message
	* `actor`: the actor which is signing the message
	* `id`: the ActivityStreams URI of the message
	"""
	def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do
	Logger.debug("Federating #{id} to #{inbox}")

	case FedSockets.publish(inbox, json) do
	:ok ->
	Logger.debug("Published via FedSocket - #{inspect(inbox)}")
	:ok

	- _ ->
	+ e ->
	+ Logger.debug("Shit broke - #{inspect(e)}")
	Logger.debug("publishing via http - #{inspect(inbox)}")
	http_publish(inbox, actor, json, params)
	end
	end

	def publish_one(%{actor_id: actor_id} = params) do
	actor = User.get_cached_by_id(actor_id)

	params
	\|> Map.delete(:actor_id)
	\|> Map.put(:actor, actor)
	\|> publish_one()
	end

	defp http_publish(inbox, actor, json, params) do
	uri = %{path: path} = URI.parse(inbox)
	digest = "SHA-256=" <> (:crypto.hash(:sha256, json) \|> Base.encode64())

	date = Pleroma.Signature.signed_date()

	signature =
	Pleroma.Signature.sign(actor, %{
	"(request-target)": "post #{path}",
	host: signature_host(uri),
	"content-length": byte_size(json),
	digest: digest,
	date: date
	})

	with {:ok, %{status: code}} when code in 200..299 <-
	result =
	HTTP.post(
	inbox,
	json,
	[
	{"Content-Type", "application/activity+json"},
	{"Date", date},
	{"signature", signature},
	{"digest", digest}
	]
	) do
	if not Map.has_key?(params, :unreachable_since) \|\| params[:unreachable_since] do
	Instances.set_reachable(inbox)
	end

	result
	else
	{_post_result, response} ->
	unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
	{:error, response}
	end
	end

	defp signature_host(%URI{port: port, scheme: scheme, host: host}) do
	if port == URI.default_port(scheme) do
	host
	else
	"#{host}:#{port}"
	end
	end

	defp should_federate?(inbox, public) do
	if public do
	true
	else
	%{host: host} = URI.parse(inbox)

	quarantined_instances =
	Config.get([:instance, :quarantined_instances], [])
	\|> Pleroma.Web.ActivityPub.MRF.subdomains_regex()

	!Pleroma.Web.ActivityPub.MRF.subdomain_match?(quarantined_instances, host)
	end
	end

	@spec recipients(User.t(), Activity.t()) :: list(User.t()) \| []
	defp recipients(actor, activity) do
	followers =
	if actor.follower_address in activity.recipients do
	User.get_external_followers(actor)
	else
	[]
	end

	fetchers =
	with %Activity{data: %{"type" => "Delete"}} <- activity,
	%Object{id: object_id} <- Object.normalize(activity),
	fetchers <- User.get_delivered_users_by_object_id(object_id),
	_ <- Delivery.delete_all_by_object_id(object_id) do
	fetchers
	else
	_ ->
	[]
	end

	Pleroma.Web.Federator.Publisher.remote_users(actor, activity) ++ followers ++ fetchers
	end

	defp get_cc_ap_ids(ap_id, recipients) do
	host = Map.get(URI.parse(ap_id), :host)

	recipients
	\|> Enum.filter(fn %User{ap_id: ap_id} -> Map.get(URI.parse(ap_id), :host) == host end)
	\|> Enum.map(& &1.ap_id)
	end

	defp maybe_use_sharedinbox(%User{shared_inbox: nil, inbox: inbox}), do: inbox
	defp maybe_use_sharedinbox(%User{shared_inbox: shared_inbox}), do: shared_inbox

	@doc """
	Determine a user inbox to use based on heuristics. These heuristics
	are based on an approximation of the ``sharedInbox`` rules in the
	[ActivityPub specification][ap-sharedinbox].

	Please do not edit this function (or its children) without reading
	the spec, as editing the code is likely to introduce some breakage
	without some familiarity.

	[ap-sharedinbox]: https://www.w3.org/TR/activitypub/#shared-inbox-delivery
	"""
	def determine_inbox(
	%Activity{data: activity_data},
	%User{inbox: inbox} = user
	) do
	to = activity_data["to"] \|\| []
	cc = activity_data["cc"] \|\| []
	type = activity_data["type"]

	cond do
	type == "Delete" ->
	maybe_use_sharedinbox(user)

	Pleroma.Constants.as_public() in to \|\| Pleroma.Constants.as_public() in cc ->
	maybe_use_sharedinbox(user)

	length(to) + length(cc) > 1 ->
	maybe_use_sharedinbox(user)

	true ->
	inbox
	end
	end

	@doc """
	Publishes an activity with BCC to all relevant peers.
	"""

	def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity)
	when is_list(bcc) and bcc != [] do
	public = is_public?(activity)
	{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)

	recipients = recipients(actor, activity)

	inboxes =
	recipients
	\|> Enum.filter(&User.ap_enabled?/1)
	\|> Enum.map(fn actor -> actor.inbox end)
	\|> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
	\|> Instances.filter_reachable()

	Repo.checkout(fn ->
	Enum.each(inboxes, fn {inbox, unreachable_since} ->
	%User{ap_id: ap_id} = Enum.find(recipients, fn actor -> actor.inbox == inbox end)

	# Get all the recipients on the same host and add them to cc. Otherwise, a remote
	# instance would only accept a first message for the first recipient and ignore the rest.
	cc = get_cc_ap_ids(ap_id, recipients)

	json =
	data
	\|> Map.put("cc", cc)
	\|> Jason.encode!()

	Pleroma.Web.Federator.Publisher.enqueue_one(__MODULE__, %{
	inbox: inbox,
	json: json,
	actor_id: actor.id,
	id: activity.data["id"],
	unreachable_since: unreachable_since
	})
	end)
	end)
	end

	@doc """
	Publishes an activity to all relevant peers.
	"""
	def publish(%User{} = actor, %Activity{} = activity) do
	public = is_public?(activity)

	if public && Config.get([:instance, :allow_relay]) do
	Logger.debug(fn -> "Relaying #{activity.data["id"]} out" end)
	Relay.publish(activity)
	end

	{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
	json = Jason.encode!(data)

	recipients(actor, activity)
	\|> Enum.filter(fn user -> User.ap_enabled?(user) end)
	\|> Enum.map(fn %User{} = user ->
	determine_inbox(activity, user)
	end)
	\|> Enum.uniq()
	\|> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
	\|> Instances.filter_reachable()
	\|> Enum.each(fn {inbox, unreachable_since} ->
	Pleroma.Web.Federator.Publisher.enqueue_one(
	__MODULE__,
	%{
	inbox: inbox,
	json: json,
	actor_id: actor.id,
	id: activity.data["id"],
	unreachable_since: unreachable_since
	}
	)
	end)
	end

	def gather_webfinger_links(%User{} = user) do
	[
	%{"rel" => "self", "type" => "application/activity+json", "href" => user.ap_id},
	%{
	"rel" => "self",
	"type" => "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
	"href" => user.ap_id
	},
	%{
	"rel" => "http://ostatus.org/schema/1.0/subscribe",
	"template" => "#{Pleroma.Web.base_url()}/ostatus_subscribe?acct={uri}"
	}
	]
	end

	def gather_nodeinfo_protocol_names, do: ["activitypub"]
	end
	diff --git a/lib/pleroma/web/fed_sockets/adapter.ex b/lib/pleroma/web/fed_sockets/adapter.ex
	index 5bd9caec9..00ab2d056 100644
	--- a/lib/pleroma/web/fed_sockets/adapter.ex
	+++ b/lib/pleroma/web/fed_sockets/adapter.ex
	@@ -1,69 +1,90 @@
	defmodule Pleroma.Web.FedSockets.Adapter do
	@moduledoc """
	A behavior both types of sockets (server and client) should implement
	- and a collection of helper functions useful to both.
	+ and a collection of helper functions useful to both
	"""

	@type adapter_state :: map()

	@doc "A synchronous fetch."
	@callback fetch(pid(), adapter_state(), term(), timeout()) :: {:ok, term()} \| {:error, term()}

	@doc "An asynchronous publish."
	@callback publish(pid(), adapter_state(), term()) :: :ok \| {:error, term()}

	alias Pleroma.Object
	alias Pleroma.Object.Containment
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ObjectView
	alias Pleroma.Web.ActivityPub.UserView
	alias Pleroma.Web.ActivityPub.Visibility
	alias Pleroma.Web.FedSockets.IngesterWorker

	- @typedoc """
	- Should be "fetch" or "publish"
	- """
	- @type common_action :: String.t()
	@type origin :: String.t()
	- @doc "Process non adapter-specific messages."
	- @spec process_message(map(), origin()) ::
	- {:reply, term()} \| :noreply \| {:error, :unknown_action}
	- def process_message(%{"action" => "publish", "data" => data}, origin) do
	+ @type fetch_id :: integer()
	+ @type waiting_fetches :: %{required(fetch_id()) => pid()}
	+ @doc "Processes incoming messages. Returns {:reply, websocket_frame, waiting_fetches} or `{:noreply, waiting_fetches}`"
	+ @spec process_message(binary() \| map(), origin(), waiting_fetches()) ::
	+ {:reply, term(), waiting_fetches()} \| {:noreply, waiting_fetches()}
	+ def process_message(message, origin, waiting_fetches) when is_binary(message) do
	+ case Jason.decode(message) do
	+ {:ok, message} -> process_message(message, origin, waiting_fetches)
	+ # 1003 indicates that an endpoint is terminating the connection
	+ # because it has received a type of data it cannot accept.
	+ {:error, decode_error} -> {:reply, {:close, 1003, Exception.message(decode_error)}}
	+ end
	+ end
	+
	+ def process_message(%{"action" => "publish", "data" => data}, origin, waiting_fetches) do
	if Containment.contain_origin(origin, data) do
	IngesterWorker.enqueue("ingest", %{"object" => data})
	end

	- :noreply
	+ {:noreply, waiting_fetches}
	end

	- def process_message(%{"action" => "fetch", "uuid" => uuid, "data" => ap_id}, _) do
	+ def process_message(%{"action" => "fetch", "uuid" => uuid, "data" => ap_id}, _, _) do
	data = %{
	"action" => "fetch_reply",
	"status" => "processed",
	"uuid" => uuid,
	"data" => represent_item(ap_id)
	}

	- {:reply, data}
	+ {:reply, {:text, Jason.encode!(data)}}
	+ end
	+
	+ def process_message(
	+ %{"action" => "fetch_reply", "uuid" => uuid, "data" => data},
	+ _,
	+ waiting_fetches
	+ ) do
	+ with {pid, waiting_fetches} when is_pid(pid) <- Map.pop(waiting_fetches, uuid) do
	+ send(pid, {:fetch_reply, uuid, data})
	+ {:noreply, waiting_fetches}
	+ else
	+ _ ->
	+ {:noreply, waiting_fetches}
	+ end
	end

	- def process_message(_, _) do
	- {:error, :unknown_action}
	+ def process_message(_, _, waiting_fetches) do
	+ {:reply, {:close, 1003, "Unknown message type."}, waiting_fetches}
	end

	defp represent_item(ap_id) do
	case User.get_by_ap_id(ap_id) do
	nil ->
	object = Object.get_cached_by_ap_id(ap_id)

	if Visibility.is_public?(object) do
	Phoenix.View.render(ObjectView, "object.json", object: object)
	else
	nil
	end

	user ->
	Phoenix.View.render(UserView, "user.json", user: user)
	end
	end
	end
	diff --git a/lib/pleroma/web/fed_sockets/adapter/cowboy.ex b/lib/pleroma/web/fed_sockets/adapter/cowboy.ex
	index a732df159..880af929c 100644
	--- a/lib/pleroma/web/fed_sockets/adapter/cowboy.ex
	+++ b/lib/pleroma/web/fed_sockets/adapter/cowboy.ex
	@@ -1,155 +1,144 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.FedSockets.Adapter.Cowboy do
	require Logger

	alias Pleroma.Web.FedSockets.Adapter
	alias Pleroma.Web.FedSockets.FedSocket
	alias Pleroma.Web.FedSockets.Registry.Value

	import HTTPSignatures, only: [validate_conn: 1, split_signature: 1]

	require Logger

	@behaviour :cowboy_websocket
	@behaviour Adapter

	@impl true
	def fetch(pid, %{last_fetch_id_ref: last_fetch_id_ref}, id, timeout) do
	fetch_id = :atomics.add_get(last_fetch_id_ref, 1, 1)
	message = %{action: :fetch, data: id, uuid: fetch_id}
	send(pid, {:send_fetch, Jason.encode!(message), fetch_id, self()})

	receive do
	{:fetch_reply, ^fetch_id, data} -> {:ok, data}
	after
	timeout -> {:error, :timeout}
	end
	end

	@impl true
	def publish(pid, _, data) do
	message = %{action: :publish, data: data}
	send(pid, {:send, Jason.encode!(message)})
	+ :ok
	end

	@impl true
	def init(req, state) do
	shake = FedSocket.shake()

	with {_, true} <- {:enabled, Pleroma.Config.get([:fed_sockets, :enabled])},
	sec_protocol <- :cowboy_req.header("sec-websocket-protocol", req, nil),
	{_, %{"(request-target)" => ^shake} = headers} <-
	{:has_request_target, :cowboy_req.headers(req)},
	{_, true} <- {:signature_validated, validate_conn(%{req_headers: headers})},
	%{"keyId" => origin} <- split_signature(headers["signature"]) do
	req =
	if is_nil(sec_protocol) do
	req
	else
	:cowboy_req.set_resp_header("sec-websocket-protocol", sec_protocol, req)
	end

	{:cowboy_websocket, req, origin, %{}}
	else
	{:has_request_target, headers} ->
	Logger.debug(fn ->
	"#{__MODULE__}: Wrong or no \"(request-target)\" header. Rejecting websocket switch. Headers:\n#{
	inspect(headers)
	}"
	end)

	:cowboy_req.reply(400, req)
	{:ok, req, state}

	{:signature_validated, false} ->
	Logger.debug(fn ->
	"#{__MODULE__}: Signature validation failed. Rejecting websocket switch."
	end)

	:cowboy_req.reply(401, req)
	{:ok, req, state}

	e ->
	Logger.debug(fn -> "#{__MODULE__}: Websocket switch failed, #{inspect(e)}" end)
	:cowboy_req.reply(500, req)
	{:ok, req, state}
	end
	end

	@registry Pleroma.Web.FedSockets.Registry

	@impl true
	def websocket_init(origin) do
	key = Pleroma.Web.FedSockets.Registry.key_from_uri(URI.parse(origin))
	# Since, unlike with gun, we don't have calls.
	# We store last fetch id in an atomic counter and use casts.
	last_fetch_id_ref = :atomics.new(1, [])
	:ok = :atomics.put(last_fetch_id_ref, 1, 0)

	case Registry.register(@registry, key, %Value{
	adapter: __MODULE__,
	adapter_state: %{last_fetch_id_ref: last_fetch_id_ref, waiting_fetches: %{}}
	}) do
	{:ok, _owner} ->
	{:ok, %{origin: origin}}

	{:error, {:already_registered, _}} ->
	{:stop, origin}
	end
	end

	@impl true
	def websocket_handle(:ping, socket_info), do: {:ok, socket_info}

	- def websocket_handle({:text, raw_message}, %{origin: origin} = state) do
	- case Jason.decode(raw_message) do
	- {:ok, message} ->
	- case message do
	- %{"action" => "fetch_reply", "uuid" => uuid, "data" => data} ->
	- with {pid, waiting_fetches} when is_pid(pid) <- Map.pop(state.waiting_fetches, uuid) do
	- send(pid, {:fetch_reply, uuid, data})
	- {:ok, %{state \| waiting_fetches: waiting_fetches}}
	- else
	- _ ->
	- {:ok, state}
	- end
	-
	- message ->
	- case Adapter.process_message(message, origin) do
	- :noreply -> {:ok, state}
	- {:reply, data} -> {:reply, {:text, Jason.encode!(data)}, state}
	- end
	- end
	+ def websocket_handle(
	+ {:text, raw_message},
	+ %{origin: origin, waiting_fetches: waiting_fetches} = state
	+ ) do
	+ case Adapter.process_message(raw_message, origin, waiting_fetches) do
	+ {:reply, frame, waiting_fetches} ->
	+ {:reply, frame, %{state \| waiting_fetches: waiting_fetches}}

	- {:error, decode_error} ->
	- exit({:malformed_message, decode_error})
	+ {:noreply, waiting_fetches} ->
	+ {:ok, %{state \| waiting_fetches: waiting_fetches}}
	end
	end

	@impl true
	def websocket_info(
	{:send_fetch, message, fetch_id, pid},
	%{waiting_fetches: waiting_fetches} = state
	) do
	waiting_fetches = Map.put(waiting_fetches, fetch_id, pid)
	{:reply, {:text, message}, %{state \| waiting_fetches: waiting_fetches}}
	end

	@impl true
	def websocket_info({:send, message}, state) do
	{:reply, {:text, message}, state}
	end

	@impl true
	def websocket_info(:close, state) do
	{:stop, state}
	end

	def websocket_info(message, state) do
	Logger.debug("#{__MODULE__} unknown message #{inspect(message)}")
	{:ok, state}
	end
	end
	diff --git a/lib/pleroma/web/fed_sockets/adapter/gun.ex b/lib/pleroma/web/fed_sockets/adapter/gun.ex
	index 51c4e6be9..dc8ae6db7 100644
	--- a/lib/pleroma/web/fed_sockets/adapter/gun.ex
	+++ b/lib/pleroma/web/fed_sockets/adapter/gun.ex
	@@ -1,240 +1,231 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.FedSockets.Adapter.Gun do
	use GenServer, restart: :temporary

	require Logger

	alias Pleroma.Web.ActivityPub.InternalFetchActor
	alias Pleroma.Web.FedSockets.Registry.Value
	alias Pleroma.Web.FedSockets.FedSocket
	alias Pleroma.Web.FedSockets.Adapter

	@behaviour Adapter

	@registry Pleroma.Web.FedSockets.Registry

	@impl true
	def fetch(pid, state, id, timeout) do
	- # TODO: refactor with atomics and encoding on the client
	- with {:ok, _} <- await_connected(pid, state) do
	- GenServer.call(pid, {:fetch, id}, timeout)
	+ with {:ok, conn_pid, last_fetch_id_ref} <- await_connected(pid, state) do
	+ fetch_id = :atomics.add_get(last_fetch_id_ref, 1, 1)
	+ message = %{action: :fetch, data: id, uuid: fetch_id}
	+ send(pid, {:register_fetch, fetch_id, self()})
	+ send_json(conn_pid, message)
	+
	+ receive do
	+ {:fetch_reply, ^fetch_id, data} -> {:ok, data}
	+ after
	+ timeout -> {:error, :timeout}
	+ end
	end
	end

	@impl true
	def publish(pid, state, data) do
	- with {:ok, conn_pid} <- await_connected(pid, state) do
	+ with {:ok, conn_pid, _} <- await_connected(pid, state) do
	send_json(conn_pid, %{action: :publish, data: data})
	end
	end

	- defp await_connected(_pid, %{conn_pid: conn_pid}), do: {:ok, conn_pid}
	+ defp await_connected(_pid, %{conn_pid: conn_pid, last_fetch_id_ref: last_fetch_id_ref}),
	+ do: {:ok, conn_pid, last_fetch_id_ref}

	defp await_connected(pid, _) do
	monitor = Process.monitor(pid)
	GenServer.cast(pid, {:await_connected, self()})

	receive do
	{:DOWN, ^monitor, _, _, {:shutdown, reason}} -> reason
	{:DOWN, ^monitor, _, _, reason} -> {:error, reason}
	- {:await_connected, ^pid, conn_pid} -> {:ok, conn_pid}
	+ {:await_connected, ^pid, conn_pid, last_fetch_id_ref} -> {:ok, conn_pid, last_fetch_id_ref}
	end
	end

	def start_link([key \| _] = opts) do
	- GenServer.start_link(__MODULE__, opts,
	- name: {:via, Registry, {@registry, key, %Value{adapter: __MODULE__, adapter_state: %{}}}}
	+ last_fetch_id_ref = :atomics.new(1, [])
	+ :ok = :atomics.put(last_fetch_id_ref, 1, 0)
	+
	+ GenServer.start_link(__MODULE__, [last_fetch_id_ref \| opts],
	+ name:
	+ {:via, Registry,
	+ {@registry, key,
	+ %Value{adapter: __MODULE__, adapter_state: %{last_fetch_id_ref: last_fetch_id_ref}}}}
	)
	end

	@impl true
	def init(opts) do
	{:ok, nil, {:continue, {:connect, opts}}}
	end

	@impl true
	- def handle_continue({:connect, [key, uri] = opts}, _) do
	+ def handle_continue({:connect, [last_fetch_id_ref, key, uri] = opts}, _) do
	case initiate_connection(uri) do
	{:ok, conn_pid} ->
	Registry.update_value(@registry, key, fn value ->
	- %{value \| adapter_state: %{conn_pid: conn_pid}}
	+ %{value \| adapter_state: Map.put(value.adapter_state, :conn_pid, conn_pid)}
	end)

	{:noreply,
	- %{conn_pid: conn_pid, waiting_fetches: %{}, last_fetch_id: 0, origin: uri, key: key}}
	+ %{
	+ conn_pid: conn_pid,
	+ waiting_fetches: %{},
	+ last_fetch_id_ref: last_fetch_id_ref,
	+ origin: uri,
	+ key: key
	+ }}

	{:error, reason} = e ->
	Logger.debug("Outgoing connection failed - #{inspect(reason)}")
	{:stop, {:shutdown, e}, opts}
	end
	end

	@impl true
	- def handle_cast({:await_connected, pid}, %{conn_pid: conn_pid} = state) do
	- send(pid, {:await_connected, self(), conn_pid})
	- {:noreply, state}
	- end
	-
	- @impl true
	- def handle_call(
	- {:fetch, data},
	- from,
	- %{
	- last_fetch_id: last_fetch_id,
	- conn_pid: conn_pid,
	- waiting_fetches: waiting_fetches
	- } = state
	+ def handle_cast(
	+ {:await_connected, pid},
	+ %{conn_pid: conn_pid, last_fetch_id_ref: last_fetch_id_ref} = state
	) do
	- last_fetch_id = last_fetch_id + 1
	- request = %{action: :fetch, data: data, uuid: last_fetch_id}
	- :ok = send_json(conn_pid, request)
	- waiting_fetches = Map.put(waiting_fetches, last_fetch_id, from)
	-
	- {:noreply,
	- %{
	- state
	- \| waiting_fetches: waiting_fetches,
	- last_fetch_id: last_fetch_id
	- }}
	+ send(pid, {:await_connected, self(), conn_pid, last_fetch_id_ref})
	+ {:noreply, state}
	end

	defp send_json(conn_pid, data) do
	:gun.ws_send(conn_pid, {:text, Jason.encode!(data)})
	end

	+ @impl true
	+ def handle_info({:register_fetch, fetch_id, pid}, %{waiting_fetches: waiting_fetches} = state) do
	+ waiting_fetches = Map.put(waiting_fetches, fetch_id, pid)
	+ {:noreply, %{state \| waiting_fetches: waiting_fetches}}
	+ end
	+
	@impl true
	def handle_info(
	{:gun_ws, _conn_pid, _ref, {:text, raw_message}},
	- %{conn_pid: conn_pid, origin: origin} = state
	+ %{conn_pid: conn_pid, origin: origin, waiting_fetches: waiting_fetches} = state
	) do
	- state =
	- case Jason.decode(raw_message) do
	- {:ok, message} ->
	- case message do
	- %{"action" => "fetch_reply", "uuid" => uuid, "data" => data} ->
	- with {{_, _} = client, waiting_fetches} <- Map.pop(state.waiting_fetches, uuid) do
	- GenServer.reply(client, {:ok, data})
	- %{state \| waiting_fetches: waiting_fetches}
	- else
	- _ ->
	- state
	- end
	-
	- message ->
	- case Adapter.process_message(message, origin) do
	- :noreply -> :noop
	- {:reply, data} -> send_json(conn_pid, data)
	- end
	-
	- state
	- end
	-
	- {:error, decode_error} ->
	- exit({:malformed_message, decode_error})
	+ waiting_fetches =
	+ case Adapter.process_message(raw_message, origin, waiting_fetches) do
	+ {:reply, frame, waiting_fetches} ->
	+ :gun.ws_send(conn_pid, frame)
	+ waiting_fetches
	+
	+ {:noreply, waiting_fetches} ->
	+ waiting_fetches
	end

	- {:noreply, state}
	+ {:noreply, %{state \| waiting_fetches: waiting_fetches}}
	end

	@impl true
	def handle_info(:close, state) do
	Logger.debug("Sending close frame !!!!!!!")
	{:close, state}
	end

	@impl true
	def handle_info({:gun_down, _pid, _prot, :closed, _}, state) do
	{:stop, :normal, state}
	end

	@impl true
	def handle_info({:gun_ws, _, _, :pong}, state) do
	{:noreply, state, :hibernate}
	end

	@impl true
	def handle_info(msg, state) do
	Logger.debug("#{__MODULE__} unhandled event #{inspect(msg)}")
	{:noreply, state}
	end

	@impl true
	def terminate(reason, state) do
	Logger.debug(
	"#{__MODULE__} terminating outgoing connection for #{inspect(state)} for #{inspect(reason)}"
	)

	{:ok, state}
	end

	@path '/api/fedsocket/v1'
	def initiate_connection(uri) do
	%{host: host, port: port} = URI.parse(uri)

	with {:ok, conn_pid} <- :gun.open(to_charlist(host), port, %{protocols: [:http]}),
	{:ok, _} <- :gun.await_up(conn_pid),
	# TODO: nodeinfo-based support detection
	# reference <- :gun.get(conn_pid, to_charlist(path)),
	# {:response, :fin, 204, _} <- :gun.await(conn_pid, reference) \|> IO.inspect(),
	# :ok <- :gun.flush(conn_pid),
	headers <- build_headers(uri),
	ref <- :gun.ws_upgrade(conn_pid, @path, headers, %{silence_pings: false}) do
	receive do
	{:gun_upgrade, ^conn_pid, ^ref, [<<"websocket">>], _} ->
	{:ok, conn_pid}

	# mes ->
	# IO.inspect(mes)
	after
	15_000 ->
	Logger.debug("Fedsocket timeout connecting to #{inspect(uri)}")
	{:error, :timeout}
	end
	else
	{:response, :nofin, 404, _} ->
	{:error, :fedsockets_not_supported}

	e ->
	Logger.debug("Fedsocket error connecting to #{inspect(uri)}")
	{:error, e}
	end
	end

	defp build_headers(uri) do
	host_for_sig = uri \|> URI.parse() \|> host_signature()

	shake = FedSocket.shake()
	digest = "SHA-256=" <> (:crypto.hash(:sha256, shake) \|> Base.encode64())
	date = Pleroma.Signature.signed_date()
	shake_size = byte_size(shake)

	signature_opts = %{
	"(request-target)": shake,
	"content-length": to_charlist("#{shake_size}"),
	date: date,
	digest: digest,
	host: host_for_sig
	}

	signature = Pleroma.Signature.sign(InternalFetchActor.get_actor(), signature_opts)

	[
	{"signature", signature},
	{"date", date},
	{"digest", digest},
	{"content-length", to_string(shake_size)},
	{"(request-target)", shake}
	]
	end

	defp host_signature(%{host: host, scheme: scheme, port: port}) do
	if port == URI.default_port(scheme) do
	host
	else
	"#{host}:#{port}"
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sat, Aug 29, 9:35 AM (1 d, 20 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1736982
Default Alt Text: (27 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions